Graph Operations

Complete API for graph data access, transformations, and manipulations.

Data Access Methods

Basic Data Retrieval

def getVertices: DataSet[Vertex[K, VV]]

Returns the vertex DataSet containing all graph vertices.

def getEdges: DataSet[Edge[K, EV]]

Returns the edge DataSet containing all graph edges.

def getVerticesAsTuple2(): DataSet[(K, VV)]

Returns vertices as tuple DataSet (vertexId, vertexValue).

def getEdgesAsTuple3(): DataSet[(K, K, EV)]

Returns edges as tuple DataSet (sourceId, targetId, edgeValue).

def getTriplets(): DataSet[Triplet[K, VV, EV]]

Returns triplets containing (srcVertexId, trgVertexId, srcVertexValue, trgVertexValue, edgeValue).

Graph Statistics

def numberOfVertices(): Long

Returns the total number of vertices in the graph.

def numberOfEdges(): Long

Returns the total number of edges in the graph.

def getVertexIds(): DataSet[K]

Returns a DataSet containing all vertex IDs.

def getEdgeIds(): DataSet[(K, K)]

Returns a DataSet containing all edge IDs as (sourceId, targetId) tuples.

Transformation Operations

Vertex Transformations

def mapVertices[NV: TypeInformation : ClassTag](mapper: MapFunction[Vertex[K, VV], NV]): Graph[K, NV, EV]

Applies a map function to transform vertex values.

Parameters:

• mapper - MapFunction that transforms Vertex[K, VV] to new value type NV

def mapVertices[NV: TypeInformation : ClassTag](fun: Vertex[K, VV] => NV): Graph[K, NV, EV]

Applies a Scala function to transform vertex values.

Parameters:

• fun - Scala function that transforms vertex to new value type

Edge Transformations

def mapEdges[NV: TypeInformation : ClassTag](mapper: MapFunction[Edge[K, EV], NV]): Graph[K, VV, NV]

Applies a map function to transform edge values.

Parameters:

• mapper - MapFunction that transforms Edge[K, EV] to new value type NV

def mapEdges[NV: TypeInformation : ClassTag](fun: Edge[K, EV] => NV): Graph[K, VV, NV]

Applies a Scala function to transform edge values.

Parameters:

• fun - Scala function that transforms edge to new value type

Translation Operations

def translateGraphIds[NEW: TypeInformation : ClassTag](translator: TranslateFunction[K, NEW]): Graph[NEW, VV, EV]

Translates vertex and edge IDs using the given TranslateFunction.

Parameters:

• translator - TranslateFunction implementing conversion from K to NEW

def translateGraphIds[NEW: TypeInformation : ClassTag](fun: (K, NEW) => NEW): Graph[NEW, VV, EV]

Translates vertex and edge IDs using a Scala function.

Parameters:

• fun - Function implementing conversion from K to NEW

def translateVertexValues[NEW: TypeInformation : ClassTag](translator: TranslateFunction[VV, NEW]): Graph[K, NEW, EV]

Translates vertex values using the given TranslateFunction.

Parameters:

• translator - TranslateFunction implementing conversion from VV to NEW

def translateVertexValues[NEW: TypeInformation : ClassTag](fun: (VV, NEW) => NEW): Graph[K, NEW, EV]

Translates vertex values using a Scala function.

Parameters:

• fun - Function implementing conversion from VV to NEW

def translateEdgeValues[NEW: TypeInformation : ClassTag](translator: TranslateFunction[EV, NEW]): Graph[K, VV, NEW]

Translates edge values using the given TranslateFunction.

Parameters:

• translator - TranslateFunction implementing conversion from EV to NEW

def translateEdgeValues[NEW: TypeInformation : ClassTag](fun: (EV, NEW) => NEW): Graph[K, VV, NEW]

Translates edge values using a Scala function.

Parameters:

• fun - Function implementing conversion from EV to NEW

Join Operations

Vertex Joins

def joinWithVertices[T: TypeInformation](inputDataSet: DataSet[(K, T)], vertexJoinFunction: VertexJoinFunction[VV, T]): Graph[K, VV, EV]

Joins the vertex DataSet with an input Tuple2 DataSet and applies a transformation function.

Parameters:

• inputDataSet - Tuple2 DataSet to join with (vertexId, inputValue)
• vertexJoinFunction - Function to transform joined values

def joinWithVertices[T: TypeInformation](inputDataSet: DataSet[(K, T)], fun: (VV, T) => VV): Graph[K, VV, EV]

Joins vertices with external data using a Scala function.

Parameters:

• inputDataSet - Tuple2 DataSet (vertexId, inputValue)
• fun - Function combining current vertex value with input value

Edge Joins

def joinWithEdges[T: TypeInformation](inputDataSet: DataSet[(K, K, T)], edgeJoinFunction: EdgeJoinFunction[EV, T]): Graph[K, VV, EV]

Joins the edge DataSet with an input DataSet on the composite key of source and target IDs.

Parameters:

• inputDataSet - Tuple3 DataSet (sourceId, targetId, inputValue)
• edgeJoinFunction - Function to transform joined edge values

def joinWithEdges[T: TypeInformation](inputDataSet: DataSet[(K, K, T)], fun: (EV, T) => EV): Graph[K, VV, EV]

Joins edges with external data using a Scala function.

def joinWithEdgesOnSource[T: TypeInformation](inputDataSet: DataSet[(K, T)], edgeJoinFunction: EdgeJoinFunction[EV, T]): Graph[K, VV, EV]

Joins edges with external data based on source vertex ID.

Parameters:

• inputDataSet - Tuple2 DataSet (sourceId, inputValue)
• edgeJoinFunction - Function to transform edge values

def joinWithEdgesOnSource[T: TypeInformation](inputDataSet: DataSet[(K, T)], fun: (EV, T) => EV): Graph[K, VV, EV]

Joins edges on source ID using a Scala function.

def joinWithEdgesOnTarget[T: TypeInformation](inputDataSet: DataSet[(K, T)], edgeJoinFunction: EdgeJoinFunction[EV, T]): Graph[K, VV, EV]

Joins edges with external data based on target vertex ID.

def joinWithEdgesOnTarget[T: TypeInformation](inputDataSet: DataSet[(K, T)], fun: (EV, T) => EV): Graph[K, VV, EV]

Joins edges on target ID using a Scala function.

Filtering Operations

Subgraph Creation

def subgraph(vertexFilter: FilterFunction[Vertex[K, VV]], edgeFilter: FilterFunction[Edge[K, EV]]): Graph[K, VV, EV]

Creates a subgraph by applying filtering predicates to both vertices and edges.

Parameters:

• vertexFilter - Predicate function for vertices
• edgeFilter - Predicate function for edges

def subgraph(vertexFilterFun: Vertex[K, VV] => Boolean, edgeFilterFun: Edge[K, EV] => Boolean): Graph[K, VV, EV]

Creates a subgraph using Scala predicate functions.

Individual Filtering

def filterOnVertices(vertexFilter: FilterFunction[Vertex[K, VV]]): Graph[K, VV, EV]

Filters the graph based on vertex predicates only.

def filterOnVertices(vertexFilterFun: Vertex[K, VV] => Boolean): Graph[K, VV, EV]

Filters vertices using a Scala predicate function.

def filterOnEdges(edgeFilter: FilterFunction[Edge[K, EV]]): Graph[K, VV, EV]

Filters the graph based on edge predicates only.

def filterOnEdges(edgeFilterFun: Edge[K, EV] => Boolean): Graph[K, VV, EV]

Filters edges using a Scala predicate function.

Graph Mutation Operations

Adding Elements

def addVertex(vertex: Vertex[K, VV]): Graph[K, VV, EV]

Adds a single vertex to the graph. If the vertex already exists, it will not be added again.

def addVertices(vertices: List[Vertex[K, VV]]): Graph[K, VV, EV]

Adds multiple vertices to the graph.

def addEdges(edges: List[Edge[K, EV]]): Graph[K, VV, EV]

Adds multiple edges to the graph. Invalid edges (with non-existing vertices) are ignored.

def addEdge(source: Vertex[K, VV], target: Vertex[K, VV], edgeValue: EV): Graph[K, VV, EV]

Adds a single edge. If source and target vertices don't exist, they will be added.

Parameters:

• source - Source vertex of the edge
• target - Target vertex of the edge
• edgeValue - Value of the edge

Removing Elements

def removeVertex(vertex: Vertex[K, VV]): Graph[K, VV, EV]

Removes the specified vertex and all its edges from the graph.

def removeVertices(vertices: List[Vertex[K, VV]]): Graph[K, VV, EV]

Removes multiple vertices and their edges from the graph.

def removeEdge(edge: Edge[K, EV]): Graph[K, VV, EV]

Removes all edges that match the given edge from the graph.

def removeEdges(edges: List[Edge[K, EV]]): Graph[K, VV, EV]

Removes multiple edges from the graph while keeping vertices intact.

Usage Examples

Basic Transformations

// Transform vertex values to uppercase
val transformedGraph = graph.mapVertices(vertex => vertex.getValue.toUpperCase)

// Transform edge values by doubling them
val doubledEdges = graph.mapEdges(edge => edge.getValue * 2)

Filtering Operations

// Create subgraph with specific vertices and edges
val filteredGraph = graph.subgraph(
  vertex => vertex.getValue.startsWith("A"),
  edge => edge.getValue > 1.0
)

// Filter only high-value edges
val highValueGraph = graph.filterOnEdges(_.getValue > 5.0)

Join Operations

// Join with external vertex data
val externalData = env.fromCollection(List((1L, "extra"), (2L, "info")))
val enrichedGraph = graph.joinWithVertices(externalData, 
  (vertexValue, extraInfo) => s"$vertexValue-$extraInfo")

// Join with edge weights
val edgeWeights = env.fromCollection(List((1L, 2L, 2.5), (2L, 3L, 1.8)))
val weightedGraph = graph.joinWithEdges(edgeWeights,
  (currentValue, weight) => currentValue * weight)