From 7f58a8443d3617c47a86c9e723c5d2cb3fcb6c23 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Sun, 29 Jun 2025 13:57:00 +0200 Subject: [PATCH 1/9] Initial top-level structure of PropertyGraph --- .../scala/org/graphframes/GraphFrame.scala | 9 + .../scala/org/graphframes/exceptions.scala | 2 + .../propertygraph/PropertyGraphFrame.scala | 45 +++ .../property/EdgePropertyGroup.scala | 317 ++++++++++++++++++ .../property/PropertyGroup.scala | 24 ++ .../property/VertexPropertyGroup.scala | 94 ++++++ 6 files changed, 491 insertions(+) create mode 100644 src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala create mode 100644 src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala create mode 100644 src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala create mode 100644 src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala diff --git a/src/main/scala/org/graphframes/GraphFrame.scala b/src/main/scala/org/graphframes/GraphFrame.scala index 9b7907488..83b15a7d0 100644 --- a/src/main/scala/org/graphframes/GraphFrame.scala +++ b/src/main/scala/org/graphframes/GraphFrame.scala @@ -749,6 +749,15 @@ object GraphFrame extends Serializable with Logging { */ val EDGE: String = "edge" + /** + * Column name representing the weight attribute of edges in a graph. + * + * This field is used to identify and represent the weight associated with edges in a + * GraphFrame. The weight generally encodes the strength or importance of the connection between + * two nodes in a graph. + */ + val WEIGHT: String = "weight" + // ============================ Constructors and converters ================================= /** diff --git a/src/main/scala/org/graphframes/exceptions.scala b/src/main/scala/org/graphframes/exceptions.scala index 3ba75c3b9..19b244cf2 100644 --- a/src/main/scala/org/graphframes/exceptions.scala +++ b/src/main/scala/org/graphframes/exceptions.scala @@ -23,3 +23,5 @@ class InvalidPatternException() extends Exception() */ class GraphFramesUnreachableException() extends Exception("This exception should not be reachable") + +class InvalidPropertyGroupException(message: String) extends Exception(message) diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala new file mode 100644 index 000000000..e128e6457 --- /dev/null +++ b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -0,0 +1,45 @@ +package org.graphframes.propertygraph +import org.graphframes.propertygraph.property.EdgePropertyGroup +import org.graphframes.propertygraph.property.VertexPropertyGroup + +/** + * A high-level abstraction for working with property graphs that simplifies interaction with the + * GraphFrames library. + * + * PropertyGraphFrame serves as a logical structure that manages collections of vertex and edge + * property groups, providing a user-friendly API for graph operations. It handles various + * internal complexities such as: + * - ID conversion and collision prevention + * - Management of directed/undirected graph representations + * - Handling of weighted/unweighted edges + * - Data consistency across different property groups + * + * The class maintains separate collections for vertex and edge properties, allowing for flexible + * graph construction while ensuring data integrity. Each property (vertex or edge) handles its + * data internally, while this class provides a simplified interface for working with the + * underlying GraphFrame structure. + * + * Example usage: + * {{{ + * val userVertices = VertexPropertyGroup("users", userDF, "userId") + * val productVertices = VertexPropertyGroup("products", productDF, "productId") + * val purchaseEdges = EdgePropertyGroup("purchases", purchaseDF, "userId", "productId") + * + * val graph = PropertyGraphFrame( + * vertexPropertyGroups = Seq(userVertices, productVertices), + * edgesPropertyGroups = Seq(purchaseEdges) + * ) + * }}} + * + * @param vertexPropertyGroups + * Sequence of vertex property groups that define the graph's vertices + * @param edgesPropertyGroups + * Sequence of edge property groups that define the graph's edges + */ +case class PropertyGraphFrame( + vertexPropertyGroups: Seq[VertexPropertyGroup], + edgesPropertyGroups: Seq[EdgePropertyGroup]) { + vertexPropertyGroups.map(pg => pg.name -> pg).toMap + edgesPropertyGroups.map(pg => pg.name -> pg).toMap + +} diff --git a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala new file mode 100644 index 000000000..064844573 --- /dev/null +++ b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala @@ -0,0 +1,317 @@ +package org.graphframes.propertygraph.property + +import org.apache.spark.sql.Column +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.functions.xxhash64 +import org.graphframes.GraphFrame +import org.graphframes.InvalidPropertyGroupException + +/** + * Represents a logical group of edges in a property graph with associated metadata and data. + * + * EdgePropertyGroup encapsulates edge data stored in a DataFrame along with metadata describing + * how to interpret the data as graph edges. Each edge group has: + * + * - A unique name identifier + * - DataFrame containing the actual edge data + * - Direction flag indicating if edges are directed or undirected + * - Column names specifying source vertex, destination vertex and edge weight columns + * + * The class provides multiple factory methods through its companion object to create edge groups + * with: + * - Custom column names for a source, destination and weight + * - Default column names ("src", "dst", "weight") + * - Automatic weight column generation (default value 1.0) + * - Specified directionality + * + * The class validates that required columns exist in the provided DataFrame on creation. Required + * columns are: + * - Source vertex column (default: "src") + * - Destination vertex column (default: "dst") + * - Weight column (default: "weight" with value 1.0) + * + * @param name + * Unique identifier for this edge property group + * @param data + * DataFrame containing the edge data with required columns + * @param isDirected + * Whether edges should be treated as directed (true) or undirected (false) + * @param srcColumnName + * Name of the source vertex column in the data + * @param dstColumnName + * Name of the destination vertex column in the data + * @param weightColumnName + * Name of the edge weight column in the data + * @note + * When edges from different groups are combined into a GraphFrame, their SRCs and DSTs are + * hashed with the group name to prevent collisions in the same way as ID of the corresponded + * vertex group is hashed. + */ +case class EdgePropertyGroup( + val name: String, + val data: DataFrame, + isDirected: Boolean, + srcColumnName: String, + dstColumnName: String, + weightColumnName: String) + extends PropertyGroup { + import EdgePropertyGroup._ + + override protected def validate(): this.type = { + if (!data.columns.contains(srcColumnName)) { + throw new InvalidPropertyGroupException( + s"source column $srcColumnName does not exist, existed columns [${data.columns.mkString(", ")}]") + } + if (!data.columns.contains(dstColumnName)) { + throw new InvalidPropertyGroupException( + s"dest column $dstColumnName does not exist, existed columns [${data.columns.mkString(", ")}]") + } + if (!data.columns.contains(weightColumnName)) { + throw new InvalidPropertyGroupException( + s"weight column $weightColumnName does not exist, existed columns [${data.columns.mkString(", ")}]") + } + this + } + + override protected[graphframes] def internalIdMapping: DataFrame = { + data + .select(col(srcColumnName).alias(EXTERNAL_ID)) + .union(data.select(col(dstColumnName).alias(EXTERNAL_ID))) + .distinct() + .withColumn(GraphFrame.ID, xxhash64(lit(name), col(EXTERNAL_ID))) + } + + override protected[graphframes] def getData(filters: Seq[Column]): DataFrame = { + val filteredData = filters.foldLeft(data)((data, filter) => data.filter(filter)) + + val baseEdges = filteredData.select( + xxhash64(lit(name), col(srcColumnName)).alias(GraphFrame.SRC), + xxhash64(lit(name), col(dstColumnName)).alias(GraphFrame.DST), + col(weightColumnName)) + + if (isDirected) { + baseEdges + } else { + baseEdges.union( + baseEdges.select( + col(GraphFrame.DST).as(GraphFrame.SRC), + col(GraphFrame.SRC).as(GraphFrame.DST), + col(weightColumnName))) + } + } +} + +object EdgePropertyGroup { + private val EXTERNAL_ID = "externalId" + + /** + * Creates an EdgePropertyGroup with fully specified parameters + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @param isDirected + * Whether edges are directed (true) or undirected (false) + * @param srcColumnName + * Name of source vertex column in data + * @param dstColumnName + * Name of destination vertex column in data + * @param weightColumnName + * Name of edge weight column in data + * @return + * A validated EdgePropertyGroup instance + */ + def apply( + name: String, + data: DataFrame, + isDirected: Boolean, + srcColumnName: String, + dstColumnName: String, + weightColumnName: String): EdgePropertyGroup = { + new EdgePropertyGroup(name, data, isDirected, srcColumnName, dstColumnName, weightColumnName) + .validate() + } + + /** + * Creates an EdgePropertyGroup with undirected edges + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @param srcColumnName + * Name of source vertex column in data + * @param dstColumnName + * Name of destination vertex column in data + * @param weightColumnName + * Name of edge weight column in data + * @return + * A validated EdgePropertyGroup instance with isDirected=false + */ + def apply( + name: String, + data: DataFrame, + srcColumnName: String, + dstColumnName: String, + weightColumnName: String): EdgePropertyGroup = { + EdgePropertyGroup( + name, + data, + isDirected = false, + srcColumnName, + dstColumnName, + weightColumnName) + } + + /** + * Creates an EdgePropertyGroup with default column names and weights + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @return + * A validated EdgePropertyGroup instance with src/dst columns, weight=1.0, isDirected=false + */ + def apply(name: String, data: DataFrame): EdgePropertyGroup = { + val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) + EdgePropertyGroup( + name, + dataWithWeight, + isDirected = false, + GraphFrame.SRC, + GraphFrame.DST, + GraphFrame.WEIGHT) + } + + /** + * Creates an EdgePropertyGroup with default column names and specified direction + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @param isDirected + * Whether edges are directed (true) or undirected (false) + * @return + * A validated EdgePropertyGroup instance with src/dst columns and weight=1.0 + */ + def apply(name: String, data: DataFrame, isDirected: Boolean): EdgePropertyGroup = { + val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) + EdgePropertyGroup( + name, + dataWithWeight, + isDirected, + GraphFrame.SRC, + GraphFrame.DST, + GraphFrame.WEIGHT) + } + + /** + * Creates an EdgePropertyGroup with specified vertex columns + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @param srcColumnName + * Name of source vertex column in data + * @param dstColumnName + * Name of destination vertex column in data + * @return + * A validated EdgePropertyGroup instance with weight=1.0, isDirected=false + */ + def apply( + name: String, + data: DataFrame, + srcColumnName: String, + dstColumnName: String): EdgePropertyGroup = { + val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) + EdgePropertyGroup( + name, + dataWithWeight, + isDirected = false, + srcColumnName, + dstColumnName, + GraphFrame.WEIGHT) + } + + /** + * Creates an EdgePropertyGroup with specified vertex columns and direction + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @param isDirected + * Whether edges are directed (true) or undirected (false) + * @param srcColumnName + * Name of source vertex column in data + * @param dstColumnName + * Name of destination vertex column in data + * @return + * A validated EdgePropertyGroup instance with weight=1.0 + */ + def apply( + name: String, + data: DataFrame, + isDirected: Boolean, + srcColumnName: String, + dstColumnName: String): EdgePropertyGroup = { + val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) + EdgePropertyGroup( + name, + dataWithWeight, + isDirected, + srcColumnName, + dstColumnName, + GraphFrame.WEIGHT) + } + + /** + * Creates an EdgePropertyGroup with a specified weight column + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @param weightColumnName + * Name of edge weight column in data + * @return + * A validated EdgePropertyGroup instance with default src/dst columns, isDirected=false + */ + def apply(name: String, data: DataFrame, weightColumnName: String): EdgePropertyGroup = { + EdgePropertyGroup( + name, + data, + isDirected = false, + GraphFrame.SRC, + GraphFrame.DST, + weightColumnName) + } + + /** + * Creates an EdgePropertyGroup with specified weight column and direction + * + * @param name + * Unique identifier for this property group + * @param data + * Underlying DataFrame containing edge data + * @param isDirected + * Whether edges are directed (true) or undirected (false) + * @param weightColumnName + * Name of edge weight column in data + * @return + * A validated EdgePropertyGroup instance with default src/dst columns + */ + def apply( + name: String, + data: DataFrame, + isDirected: Boolean, + weightColumnName: String): EdgePropertyGroup = { + EdgePropertyGroup(name, data, isDirected, GraphFrame.SRC, GraphFrame.DST, weightColumnName) + } +} diff --git a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala new file mode 100644 index 000000000..91b50f809 --- /dev/null +++ b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala @@ -0,0 +1,24 @@ +package org.graphframes.propertygraph.property + +import org.apache.spark.sql.Column +import org.apache.spark.sql.DataFrame + +trait PropertyGroup { + val name: String + val data: DataFrame + protected def validate(): this.type + + /** + * Maintains a mapping between external IDs and internal hashed IDs used in GraphFrame + * conversion. + * + * When converting multiple edge groups to a GraphFrame, we need to ensure there are no + * collisions between source/destination vertices from different groups. This is achieved by: + * 1. Creating a hash of the vertex IDs combined with group name + * 2. Using these hashed values instead of original edge IDs in the GraphFrame + * 3. Storing this mapping internally to enable conversion back to original IDs + */ + protected[graphframes] def internalIdMapping: DataFrame + protected[graphframes] def getData(): DataFrame = getData(Seq.empty[Column]) + protected[graphframes] def getData(filters: Seq[Column]): DataFrame +} diff --git a/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala new file mode 100644 index 000000000..a34b024c7 --- /dev/null +++ b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala @@ -0,0 +1,94 @@ +package org.graphframes.propertygraph.property + +import org.apache.spark.sql.Column +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.functions.xxhash64 +import org.graphframes.GraphFrame +import org.graphframes.InvalidPropertyGroupException + +/** + * Represents a logical group of vertices in a property graph with associated data and + * identification. + * + * A VertexPropertyGroup is used to organize and manage vertices that share common characteristics + * or belong to the same logical group within a property graph. Each group maintains its own data + * in the form of a DataFrame and uses a primary key column for unique vertex identification. + * + * The class provides two ways to create a vertex property group: + * 1. With a specified primary key column: + * {{{ + * VertexPropertyGroup("users", userDataFrame, "userId") + * }}} + * 2. With the default primary key column ("id"): + * {{{ + * VertexPropertyGroup("users", userDataFrame) + * }}} + * + * @param name + * The unique identifier for this vertex property group + * @param data + * The DataFrame containing the vertex data + * @param primaryKeyColumn + * The column name used to uniquely identify vertices in this group + * @note + * When vertices from different groups are combined into a GraphFrame, their IDs are hashed with + * the group name to prevent collisions. + */ +case class VertexPropertyGroup( + val name: String, + val data: DataFrame, + val primaryKeyColumn: String) + extends PropertyGroup { + import VertexPropertyGroup._ + + override protected def validate(): this.type = { + if (!data.columns.contains(primaryKeyColumn)) { + throw new InvalidPropertyGroupException( + s"source column $primaryKeyColumn does not exist, existed columns [${data.columns.mkString(", ")}]") + } + this + } + + override protected[graphframes] def internalIdMapping: DataFrame = data + .select(col(primaryKeyColumn).alias(EXTERNAL_ID)) + .withColumn(GraphFrame.ID, xxhash64(lit(name), col(EXTERNAL_ID))) + + override protected[graphframes] def getData(filters: Seq[Column]): DataFrame = { + val filteredData = filters.foldLeft(data)((data, filter) => data.filter(filter)) + filteredData.select(xxhash64(lit(name), col(primaryKeyColumn)).alias(GraphFrame.ID)) + } +} + +object VertexPropertyGroup { + private val EXTERNAL_ID = "externalId" + + /** + * Creates a new VertexPropertyGroup with a specified primary key column. + * + * @param name + * Name of the vertex property group + * @param data + * DataFrame containing vertex data + * @param primaryKeyColumn + * Name of the column to be used as a primary key for vertex identification + * @return + * A validated VertexPropertyGroup instance + */ + def apply(name: String, data: DataFrame, primaryKeyColumn: String): VertexPropertyGroup = + new VertexPropertyGroup(name, data, primaryKeyColumn).validate() + + /** + * Creates a new VertexPropertyGroup using default a primary key column name. + * + * @param name + * Name of the vertex property group + * @param data + * DataFrame containing vertex data + * @return + * A validated VertexPropertyGroup instance + */ + def apply(name: String, data: DataFrame): VertexPropertyGroup = + new VertexPropertyGroup(name, data, GraphFrame.ID) +} From f6ed5336819aa9ca18015c596c366e0126da8d65 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Sun, 29 Jun 2025 21:30:08 +0200 Subject: [PATCH 2/9] Missing commit --- .../propertygraph/PropertyGraphFrame.scala | 29 +++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index e128e6457..d31b471b8 100644 --- a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -1,4 +1,7 @@ package org.graphframes.propertygraph + +import org.apache.spark.sql.Column +import org.graphframes.GraphFrame import org.graphframes.propertygraph.property.EdgePropertyGroup import org.graphframes.propertygraph.property.VertexPropertyGroup @@ -39,7 +42,29 @@ import org.graphframes.propertygraph.property.VertexPropertyGroup case class PropertyGraphFrame( vertexPropertyGroups: Seq[VertexPropertyGroup], edgesPropertyGroups: Seq[EdgePropertyGroup]) { - vertexPropertyGroups.map(pg => pg.name -> pg).toMap - edgesPropertyGroups.map(pg => pg.name -> pg).toMap + lazy private val vertexGroups: Map[String, VertexPropertyGroup] = + vertexPropertyGroups.map(pg => pg.name -> pg).toMap + lazy private val edgeGroups: Map[String, EdgePropertyGroup] = + edgesPropertyGroups.map(pg => pg.name -> pg).toMap + + def toGraphFrame( + vertexPropertyGroups: Seq[String], + edgePropertyGroups: Seq[String], + edgeFilters: Seq[Column], + vertexFilters: Seq[Column]): GraphFrame = { + vertexPropertyGroups.foreach(name => + require(vertexGroups.contains(name), s"Vertex property group $name does not exist")) + edgePropertyGroups.foreach(name => + require(edgeGroups.contains(name), s"Edge property group $name does not exist")) + + val vertices = vertexPropertyGroups + .map(name => vertexGroups(name).getData(vertexFilters)) + .reduce(_ union _) + + val edges = edgePropertyGroups + .map(name => edgeGroups(name).getData(edgeFilters)) + .reduce(_ union _) + GraphFrame(vertices, edges) + } } From d49adeab641913c4cfd8019e8045a37da2830d80 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Wed, 2 Jul 2025 18:11:43 +0200 Subject: [PATCH 3/9] WIP --- .../propertygraph/PropertyGraphFrame.scala | 8 +- .../property/EdgePropertyGroup.scala | 245 +++--------------- .../property/PropertyGroup.scala | 5 +- .../property/VertexPropertyGroup.scala | 11 +- 4 files changed, 47 insertions(+), 222 deletions(-) diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index d31b471b8..7a43ba28d 100644 --- a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -50,19 +50,19 @@ case class PropertyGraphFrame( def toGraphFrame( vertexPropertyGroups: Seq[String], edgePropertyGroups: Seq[String], - edgeFilters: Seq[Column], - vertexFilters: Seq[Column]): GraphFrame = { + edgeGroupFilters: Map[String, Column], + vertexGroupFilters: Map[String, Column]): GraphFrame = { vertexPropertyGroups.foreach(name => require(vertexGroups.contains(name), s"Vertex property group $name does not exist")) edgePropertyGroups.foreach(name => require(edgeGroups.contains(name), s"Edge property group $name does not exist")) val vertices = vertexPropertyGroups - .map(name => vertexGroups(name).getData(vertexFilters)) + .map(name => vertexGroups(name).getData(vertexGroupFilters(name))) .reduce(_ union _) val edges = edgePropertyGroups - .map(name => edgeGroups(name).getData(edgeFilters)) + .map(name => edgeGroups(name).getData(edgeGroupFilters(name))) .reduce(_ union _) GraphFrame(vertices, edges) diff --git a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala index 064844573..e8723de32 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala @@ -3,8 +3,9 @@ package org.graphframes.propertygraph.property import org.apache.spark.sql.Column import org.apache.spark.sql.DataFrame import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.concat import org.apache.spark.sql.functions.lit -import org.apache.spark.sql.functions.xxhash64 +import org.apache.spark.sql.functions.sha2 import org.graphframes.GraphFrame import org.graphframes.InvalidPropertyGroupException @@ -16,26 +17,24 @@ import org.graphframes.InvalidPropertyGroupException * * - A unique name identifier * - DataFrame containing the actual edge data + * - Names of source and destination vertex property groups * - Direction flag indicating if edges are directed or undirected * - Column names specifying source vertex, destination vertex and edge weight columns * - * The class provides multiple factory methods through its companion object to create edge groups - * with: - * - Custom column names for a source, destination and weight - * - Default column names ("src", "dst", "weight") - * - Automatic weight column generation (default value 1.0) - * - Specified directionality - * * The class validates that required columns exist in the provided DataFrame on creation. Required * columns are: - * - Source vertex column (default: "src") - * - Destination vertex column (default: "dst") - * - Weight column (default: "weight" with value 1.0) + * - Source vertex column + * - Destination vertex column + * - Weight column * * @param name * Unique identifier for this edge property group * @param data * DataFrame containing the edge data with required columns + * @param srcPropertyGroupName + * Name of the source vertex property group + * @param dstPropertyGroupName + * Name of the destination vertex property group * @param isDirected * Whether edges should be treated as directed (true) or undirected (false) * @param srcColumnName @@ -52,6 +51,8 @@ import org.graphframes.InvalidPropertyGroupException case class EdgePropertyGroup( val name: String, val data: DataFrame, + srcPropertyGroupName: String, + dstPropertyGroupName: String, isDirected: Boolean, srcColumnName: String, dstColumnName: String, @@ -75,20 +76,30 @@ case class EdgePropertyGroup( this } + private val hashSrcEdge: Column = + concat(lit(srcPropertyGroupName), sha2(col(srcColumnName), 256)) + private val hashDstEdge: Column = + concat(lit(dstPropertyGroupName), sha2(col(dstColumnName), 256)) + override protected[graphframes] def internalIdMapping: DataFrame = { data - .select(col(srcColumnName).alias(EXTERNAL_ID)) - .union(data.select(col(dstColumnName).alias(EXTERNAL_ID))) + .select(col(srcColumnName)) + .distinct() + .select(col(srcColumnName).alias(EXTERNAL_ID), hashSrcEdge.alias(INTERNAL_ID)) + .union( + data + .select(col(dstColumnName)) + .distinct() + .select(col(dstColumnName).alias(EXTERNAL_ID), hashDstEdge.alias(INTERNAL_ID))) .distinct() - .withColumn(GraphFrame.ID, xxhash64(lit(name), col(EXTERNAL_ID))) } - override protected[graphframes] def getData(filters: Seq[Column]): DataFrame = { - val filteredData = filters.foldLeft(data)((data, filter) => data.filter(filter)) + override protected[graphframes] def getData(filter: Column): DataFrame = { + val filteredData = data.filter(filter) val baseEdges = filteredData.select( - xxhash64(lit(name), col(srcColumnName)).alias(GraphFrame.SRC), - xxhash64(lit(name), col(dstColumnName)).alias(GraphFrame.DST), + hashSrcEdge.alias(GraphFrame.SRC), + hashDstEdge.alias(GraphFrame.DST), col(weightColumnName)) if (isDirected) { @@ -105,213 +116,25 @@ case class EdgePropertyGroup( object EdgePropertyGroup { private val EXTERNAL_ID = "externalId" + private val INTERNAL_ID = "internalId" - /** - * Creates an EdgePropertyGroup with fully specified parameters - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @param isDirected - * Whether edges are directed (true) or undirected (false) - * @param srcColumnName - * Name of source vertex column in data - * @param dstColumnName - * Name of destination vertex column in data - * @param weightColumnName - * Name of edge weight column in data - * @return - * A validated EdgePropertyGroup instance - */ def apply( name: String, data: DataFrame, + srcPropertyGroup: VertexPropertyGroup, + dstPropertyGroup: VertexPropertyGroup, isDirected: Boolean, srcColumnName: String, dstColumnName: String, weightColumnName: String): EdgePropertyGroup = { - new EdgePropertyGroup(name, data, isDirected, srcColumnName, dstColumnName, weightColumnName) - .validate() - } - - /** - * Creates an EdgePropertyGroup with undirected edges - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @param srcColumnName - * Name of source vertex column in data - * @param dstColumnName - * Name of destination vertex column in data - * @param weightColumnName - * Name of edge weight column in data - * @return - * A validated EdgePropertyGroup instance with isDirected=false - */ - def apply( - name: String, - data: DataFrame, - srcColumnName: String, - dstColumnName: String, - weightColumnName: String): EdgePropertyGroup = { EdgePropertyGroup( name, data, - isDirected = false, - srcColumnName, - dstColumnName, - weightColumnName) - } - - /** - * Creates an EdgePropertyGroup with default column names and weights - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @return - * A validated EdgePropertyGroup instance with src/dst columns, weight=1.0, isDirected=false - */ - def apply(name: String, data: DataFrame): EdgePropertyGroup = { - val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) - EdgePropertyGroup( - name, - dataWithWeight, - isDirected = false, - GraphFrame.SRC, - GraphFrame.DST, - GraphFrame.WEIGHT) - } - - /** - * Creates an EdgePropertyGroup with default column names and specified direction - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @param isDirected - * Whether edges are directed (true) or undirected (false) - * @return - * A validated EdgePropertyGroup instance with src/dst columns and weight=1.0 - */ - def apply(name: String, data: DataFrame, isDirected: Boolean): EdgePropertyGroup = { - val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) - EdgePropertyGroup( - name, - dataWithWeight, - isDirected, - GraphFrame.SRC, - GraphFrame.DST, - GraphFrame.WEIGHT) - } - - /** - * Creates an EdgePropertyGroup with specified vertex columns - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @param srcColumnName - * Name of source vertex column in data - * @param dstColumnName - * Name of destination vertex column in data - * @return - * A validated EdgePropertyGroup instance with weight=1.0, isDirected=false - */ - def apply( - name: String, - data: DataFrame, - srcColumnName: String, - dstColumnName: String): EdgePropertyGroup = { - val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) - EdgePropertyGroup( - name, - dataWithWeight, - isDirected = false, - srcColumnName, - dstColumnName, - GraphFrame.WEIGHT) - } - - /** - * Creates an EdgePropertyGroup with specified vertex columns and direction - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @param isDirected - * Whether edges are directed (true) or undirected (false) - * @param srcColumnName - * Name of source vertex column in data - * @param dstColumnName - * Name of destination vertex column in data - * @return - * A validated EdgePropertyGroup instance with weight=1.0 - */ - def apply( - name: String, - data: DataFrame, - isDirected: Boolean, - srcColumnName: String, - dstColumnName: String): EdgePropertyGroup = { - val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, lit(1.0)) - EdgePropertyGroup( - name, - dataWithWeight, + srcPropertyGroup.name, + dstPropertyGroup.name, isDirected, srcColumnName, dstColumnName, - GraphFrame.WEIGHT) - } - - /** - * Creates an EdgePropertyGroup with a specified weight column - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @param weightColumnName - * Name of edge weight column in data - * @return - * A validated EdgePropertyGroup instance with default src/dst columns, isDirected=false - */ - def apply(name: String, data: DataFrame, weightColumnName: String): EdgePropertyGroup = { - EdgePropertyGroup( - name, - data, - isDirected = false, - GraphFrame.SRC, - GraphFrame.DST, weightColumnName) } - - /** - * Creates an EdgePropertyGroup with specified weight column and direction - * - * @param name - * Unique identifier for this property group - * @param data - * Underlying DataFrame containing edge data - * @param isDirected - * Whether edges are directed (true) or undirected (false) - * @param weightColumnName - * Name of edge weight column in data - * @return - * A validated EdgePropertyGroup instance with default src/dst columns - */ - def apply( - name: String, - data: DataFrame, - isDirected: Boolean, - weightColumnName: String): EdgePropertyGroup = { - EdgePropertyGroup(name, data, isDirected, GraphFrame.SRC, GraphFrame.DST, weightColumnName) - } } diff --git a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala index 91b50f809..bd8ffbaee 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala @@ -2,6 +2,7 @@ package org.graphframes.propertygraph.property import org.apache.spark.sql.Column import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.functions.lit trait PropertyGroup { val name: String @@ -19,6 +20,6 @@ trait PropertyGroup { * 3. Storing this mapping internally to enable conversion back to original IDs */ protected[graphframes] def internalIdMapping: DataFrame - protected[graphframes] def getData(): DataFrame = getData(Seq.empty[Column]) - protected[graphframes] def getData(filters: Seq[Column]): DataFrame + protected[graphframes] def getData: DataFrame = getData(lit(true)) + protected[graphframes] def getData(filter: Column): DataFrame } diff --git a/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala index a34b024c7..e65d45821 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala @@ -3,8 +3,9 @@ package org.graphframes.propertygraph.property import org.apache.spark.sql.Column import org.apache.spark.sql.DataFrame import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.concat import org.apache.spark.sql.functions.lit -import org.apache.spark.sql.functions.xxhash64 +import org.apache.spark.sql.functions.sha2 import org.graphframes.GraphFrame import org.graphframes.InvalidPropertyGroupException @@ -53,11 +54,11 @@ case class VertexPropertyGroup( override protected[graphframes] def internalIdMapping: DataFrame = data .select(col(primaryKeyColumn).alias(EXTERNAL_ID)) - .withColumn(GraphFrame.ID, xxhash64(lit(name), col(EXTERNAL_ID))) + .withColumn(GraphFrame.ID, concat(lit(name), sha2(col(EXTERNAL_ID), 256))) - override protected[graphframes] def getData(filters: Seq[Column]): DataFrame = { - val filteredData = filters.foldLeft(data)((data, filter) => data.filter(filter)) - filteredData.select(xxhash64(lit(name), col(primaryKeyColumn)).alias(GraphFrame.ID)) + override protected[graphframes] def getData(filter: Column): DataFrame = { + val filteredData = data.filter(filter) + filteredData.select(concat(lit(name), sha2(col(primaryKeyColumn), 256)).alias(GraphFrame.ID)) } } From 30dfa4be4d4c8b7cee31a4b0b795a26ca6cae715 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Fri, 4 Jul 2025 05:35:17 +0200 Subject: [PATCH 4/9] updates from comments --- .gitignore | 1 + .../propertygraph/PropertyGraphFrame.scala | 30 +++++++++++++++++++ .../property/EdgePropertyGroup.scala | 25 ++++++++++++++-- 3 files changed, 54 insertions(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index 196465193..fa78ceed9 100644 --- a/.gitignore +++ b/.gitignore @@ -56,3 +56,4 @@ tmp/* # db-connect targets graphframes-connect-databricks/* +/workspace/ diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index 7a43ba28d..bf34754c8 100644 --- a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -47,6 +47,36 @@ case class PropertyGraphFrame( lazy private val edgeGroups: Map[String, EdgePropertyGroup] = edgesPropertyGroups.map(pg => pg.name -> pg).toMap + /** + * Converts a heterogeneous property graph into a unified GraphFrame representation. + * + * This method transforms a property graph that may contain multiple vertex types and both + * directed and undirected edges into a single GraphFrame object where all vertices and edges + * share the same schema. The conversion process handles: + * + * - Internal ID generation and collision prevention by hashing vertex/edge IDs with their + * group names + * - Merging of different vertex types into a unified vertex DataFrame + * - Conversion of directed/undirected edge relationships into a consistent edge DataFrame + * - Filtering of vertices and edges based on provided predicates + * + * The method allows selecting a subset of property groups and applying filters to control which + * data is included in the final GraphFrame. + * + * @param vertexPropertyGroups + * Sequence of vertex property group names to include in the GraphFrame + * @param edgePropertyGroups + * Sequence of edge property group names to include in the GraphFrame + * @param edgeGroupFilters + * Map of edge property group names to filter predicates (Column expressions) + * @param vertexGroupFilters + * Map of vertex property group names to filter predicates (Column expressions) + * @return + * A GraphFrame containing the unified representation of the selected and filtered property + * groups + * @throws IllegalArgumentException + * if any specified property group name doesn't exist + */ def toGraphFrame( vertexPropertyGroups: Seq[String], edgePropertyGroups: Seq[String], diff --git a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala index e8723de32..1d678c7ca 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala @@ -100,7 +100,7 @@ case class EdgePropertyGroup( val baseEdges = filteredData.select( hashSrcEdge.alias(GraphFrame.SRC), hashDstEdge.alias(GraphFrame.DST), - col(weightColumnName)) + col(weightColumnName).alias(GraphFrame.WEIGHT)) if (isDirected) { baseEdges @@ -109,7 +109,7 @@ case class EdgePropertyGroup( baseEdges.select( col(GraphFrame.DST).as(GraphFrame.SRC), col(GraphFrame.SRC).as(GraphFrame.DST), - col(weightColumnName))) + col(weightColumnName).alias(GraphFrame.WEIGHT))) } } } @@ -137,4 +137,25 @@ object EdgePropertyGroup { dstColumnName, weightColumnName) } + + def apply( + name: String, + data: DataFrame, + srcPropertyGroup: VertexPropertyGroup, + dstPropertyGroup: VertexPropertyGroup, + isDirected: Boolean, + srcColumnName: String, + dstColumnName: String, + weightColumn: Column): EdgePropertyGroup = { + val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, weightColumn) + EdgePropertyGroup( + name, + dataWithWeight, + srcPropertyGroup, + dstPropertyGroup, + isDirected, + srcColumnName, + dstColumnName, + GraphFrame.WEIGHT) + } } From 3b8c604b2908799552ad1f29fab92905fefea951 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Sat, 12 Jul 2025 07:55:27 +0200 Subject: [PATCH 5/9] From comments --- .../property/EdgePropertyGroup.scala | 64 ++++++++----------- .../property/PropertyGroup.scala | 1 - .../property/VertexPropertyGroup.scala | 2 +- 3 files changed, 28 insertions(+), 39 deletions(-) diff --git a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala index 1d678c7ca..f48d25858 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala @@ -6,6 +6,7 @@ import org.apache.spark.sql.functions.col import org.apache.spark.sql.functions.concat import org.apache.spark.sql.functions.lit import org.apache.spark.sql.functions.sha2 +import org.apache.spark.sql.types._ import org.graphframes.GraphFrame import org.graphframes.InvalidPropertyGroupException @@ -17,7 +18,7 @@ import org.graphframes.InvalidPropertyGroupException * * - A unique name identifier * - DataFrame containing the actual edge data - * - Names of source and destination vertex property groups + * - Source and destination vertex property groups * - Direction flag indicating if edges are directed or undirected * - Column names specifying source vertex, destination vertex and edge weight columns * @@ -31,10 +32,10 @@ import org.graphframes.InvalidPropertyGroupException * Unique identifier for this edge property group * @param data * DataFrame containing the edge data with required columns - * @param srcPropertyGroupName - * Name of the source vertex property group - * @param dstPropertyGroupName - * Name of the destination vertex property group + * @param srcPropertyGroup + * Source vertex property group + * @param dstPropertyGroup + * Destination vertex property group * @param isDirected * Whether edges should be treated as directed (true) or undirected (false) * @param srcColumnName @@ -48,17 +49,16 @@ import org.graphframes.InvalidPropertyGroupException * hashed with the group name to prevent collisions in the same way as ID of the corresponded * vertex group is hashed. */ -case class EdgePropertyGroup( - val name: String, - val data: DataFrame, - srcPropertyGroupName: String, - dstPropertyGroupName: String, +case class EdgePropertyGroup private ( + name: String, + data: DataFrame, + srcPropertyGroup: VertexPropertyGroup, + dstPropertyGroup: VertexPropertyGroup, isDirected: Boolean, srcColumnName: String, dstColumnName: String, weightColumnName: String) extends PropertyGroup { - import EdgePropertyGroup._ override protected def validate(): this.type = { if (!data.columns.contains(srcColumnName)) { @@ -73,26 +73,19 @@ case class EdgePropertyGroup( throw new InvalidPropertyGroupException( s"weight column $weightColumnName does not exist, existed columns [${data.columns.mkString(", ")}]") } + val weightColumnType = data.schema(weightColumnName).dataType + if (!weightColumnType.isInstanceOf[NumericType]) { + throw new InvalidPropertyGroupException( + s"weight column $weightColumnName must be numeric type, but was $weightColumnType") + } + this } - private val hashSrcEdge: Column = - concat(lit(srcPropertyGroupName), sha2(col(srcColumnName), 256)) - private val hashDstEdge: Column = - concat(lit(dstPropertyGroupName), sha2(col(dstColumnName), 256)) - - override protected[graphframes] def internalIdMapping: DataFrame = { - data - .select(col(srcColumnName)) - .distinct() - .select(col(srcColumnName).alias(EXTERNAL_ID), hashSrcEdge.alias(INTERNAL_ID)) - .union( - data - .select(col(dstColumnName)) - .distinct() - .select(col(dstColumnName).alias(EXTERNAL_ID), hashDstEdge.alias(INTERNAL_ID))) - .distinct() - } + private def hashSrcEdge: Column = + concat(lit(srcPropertyGroup.name), sha2(col(srcColumnName), 256)) + private def hashDstEdge: Column = + concat(lit(dstPropertyGroup.name), sha2(col(dstColumnName), 256)) override protected[graphframes] def getData(filter: Column): DataFrame = { val filteredData = data.filter(filter) @@ -109,15 +102,12 @@ case class EdgePropertyGroup( baseEdges.select( col(GraphFrame.DST).as(GraphFrame.SRC), col(GraphFrame.SRC).as(GraphFrame.DST), - col(weightColumnName).alias(GraphFrame.WEIGHT))) + col(GraphFrame.WEIGHT).alias(GraphFrame.WEIGHT))) } } } object EdgePropertyGroup { - private val EXTERNAL_ID = "externalId" - private val INTERNAL_ID = "internalId" - def apply( name: String, data: DataFrame, @@ -127,15 +117,15 @@ object EdgePropertyGroup { srcColumnName: String, dstColumnName: String, weightColumnName: String): EdgePropertyGroup = { - EdgePropertyGroup( + new EdgePropertyGroup( name, data, - srcPropertyGroup.name, - dstPropertyGroup.name, + srcPropertyGroup, + dstPropertyGroup, isDirected, srcColumnName, dstColumnName, - weightColumnName) + weightColumnName).validate() } def apply( @@ -148,7 +138,7 @@ object EdgePropertyGroup { dstColumnName: String, weightColumn: Column): EdgePropertyGroup = { val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, weightColumn) - EdgePropertyGroup( + apply( name, dataWithWeight, srcPropertyGroup, diff --git a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala index bd8ffbaee..6378948d1 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala @@ -19,7 +19,6 @@ trait PropertyGroup { * 2. Using these hashed values instead of original edge IDs in the GraphFrame * 3. Storing this mapping internally to enable conversion back to original IDs */ - protected[graphframes] def internalIdMapping: DataFrame protected[graphframes] def getData: DataFrame = getData(lit(true)) protected[graphframes] def getData(filter: Column): DataFrame } diff --git a/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala index e65d45821..2f73903b0 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala @@ -52,7 +52,7 @@ case class VertexPropertyGroup( this } - override protected[graphframes] def internalIdMapping: DataFrame = data + private[graphframes] def internalIdMapping: DataFrame = data .select(col(primaryKeyColumn).alias(EXTERNAL_ID)) .withColumn(GraphFrame.ID, concat(lit(name), sha2(col(EXTERNAL_ID), 256))) From b7d907663edd58081da536421d977af240ae2184 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Sat, 12 Jul 2025 15:36:03 +0200 Subject: [PATCH 6/9] Drop outdated docstrings for now --- .../propertygraph/PropertyGraphFrame.scala | 12 ------------ .../propertygraph/property/PropertyGroup.scala | 10 ---------- 2 files changed, 22 deletions(-) diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index bf34754c8..02afbdc27 100644 --- a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -22,18 +22,6 @@ import org.graphframes.propertygraph.property.VertexPropertyGroup * data internally, while this class provides a simplified interface for working with the * underlying GraphFrame structure. * - * Example usage: - * {{{ - * val userVertices = VertexPropertyGroup("users", userDF, "userId") - * val productVertices = VertexPropertyGroup("products", productDF, "productId") - * val purchaseEdges = EdgePropertyGroup("purchases", purchaseDF, "userId", "productId") - * - * val graph = PropertyGraphFrame( - * vertexPropertyGroups = Seq(userVertices, productVertices), - * edgesPropertyGroups = Seq(purchaseEdges) - * ) - * }}} - * * @param vertexPropertyGroups * Sequence of vertex property groups that define the graph's vertices * @param edgesPropertyGroups diff --git a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala index 6378948d1..173d5c040 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala @@ -9,16 +9,6 @@ trait PropertyGroup { val data: DataFrame protected def validate(): this.type - /** - * Maintains a mapping between external IDs and internal hashed IDs used in GraphFrame - * conversion. - * - * When converting multiple edge groups to a GraphFrame, we need to ensure there are no - * collisions between source/destination vertices from different groups. This is achieved by: - * 1. Creating a hash of the vertex IDs combined with group name - * 2. Using these hashed values instead of original edge IDs in the GraphFrame - * 3. Storing this mapping internally to enable conversion back to original IDs - */ protected[graphframes] def getData: DataFrame = getData(lit(true)) protected[graphframes] def getData(filter: Column): DataFrame } From 14fd2ba5807b1023ba2145167cb114093e7f1bdc Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Sun, 13 Jul 2025 09:36:14 +0200 Subject: [PATCH 7/9] Add a projection method --- .../propertygraph/PropertyGraphFrame.scala | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index 02afbdc27..2ff0d734a 100644 --- a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -1,6 +1,8 @@ package org.graphframes.propertygraph import org.apache.spark.sql.Column +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.lit import org.graphframes.GraphFrame import org.graphframes.propertygraph.property.EdgePropertyGroup import org.graphframes.propertygraph.property.VertexPropertyGroup @@ -85,4 +87,52 @@ case class PropertyGraphFrame( GraphFrame(vertices, edges) } + + /** + * Projects a bipartite graph onto one of its parts, creating edges between vertices that share + * neighbors in the other part. Drops the property group used for projection through and returns + * a new property graph. + * + * @param leftBiGraphPart + * Name of the vertex property group to project onto + * @param rightBiGraphPart + * Name of the vertex property group to project through + * @param edgeGroup + * Name of the edge property group connecting the two parts + * @return + * A new PropertyGraphFrame containing the projected graph + */ + def projectionBy( + leftBiGraphPart: String, + rightBiGraphPart: String, + edgeGroup: String): PropertyGraphFrame = { + require( + edgeGroups(edgeGroup).srcPropertyGroup.name == leftBiGraphPart, + s"Edge Property Group should have $leftBiGraphPart source group but has ${edgeGroups(edgeGroup).srcPropertyGroup.name}") + require( + edgeGroups(edgeGroup).dstPropertyGroup.name == rightBiGraphPart, + s"Edge Property Group should have $rightBiGraphPart destination group but has ${edgeGroups(edgeGroup).dstPropertyGroup.name}") + val keptVPropertyGroups = vertexPropertyGroups.filterNot(g => g.name == rightBiGraphPart) + val keptEPropertyGroups = edgesPropertyGroups.filterNot(g => g.name == edgeGroup) + val oldEdgesData = edgeGroups(edgeGroup).data + + // Create new edges by joining vertices through their common neighbors + val projectedEdges = oldEdgesData + .as("e1") + .join(oldEdgesData.as("e2"), "e1.dst = e2.dst") + .where("e1.src < e2.src") + .select(col("e1.src").alias(GraphFrame.SRC), col("e2.src").alias(GraphFrame.DST)) + + val newEdgeGroup = EdgePropertyGroup( + name = s"projected_$edgeGroup", + data = projectedEdges, + srcPropertyGroup = vertexGroups(leftBiGraphPart), + dstPropertyGroup = vertexGroups(leftBiGraphPart), + isDirected = false, + srcColumnName = GraphFrame.SRC, + dstColumnName = GraphFrame.DST, + weightColumn = lit(1.0)) + + PropertyGraphFrame(keptVPropertyGroups, keptEPropertyGroups :+ newEdgeGroup) + } } From 3068a3e292f879f8da385fb49a7d665e94aae894 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Sun, 13 Jul 2025 10:22:34 +0200 Subject: [PATCH 8/9] Add tests --- .../propertygraph/PropertyGraphFrame.scala | 2 +- .../property/EdgePropertyGroup.scala | 4 +- .../property/VertexPropertyGroup.scala | 5 +- .../PropertyGraphFrameTest.scala | 174 ++++++++++++++++++ 4 files changed, 180 insertions(+), 5 deletions(-) create mode 100644 src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index 2ff0d734a..2b230578a 100644 --- a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -119,7 +119,7 @@ case class PropertyGraphFrame( // Create new edges by joining vertices through their common neighbors val projectedEdges = oldEdgesData .as("e1") - .join(oldEdgesData.as("e2"), "e1.dst = e2.dst") + .join(oldEdgesData.as("e2"), col("e1.dst") === col("e2.dst")) .where("e1.src < e2.src") .select(col("e1.src").alias(GraphFrame.SRC), col("e2.src").alias(GraphFrame.DST)) diff --git a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala index f48d25858..0bd8881ac 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala @@ -83,9 +83,9 @@ case class EdgePropertyGroup private ( } private def hashSrcEdge: Column = - concat(lit(srcPropertyGroup.name), sha2(col(srcColumnName), 256)) + concat(lit(srcPropertyGroup.name), sha2(col(srcColumnName).cast("string"), 256)) private def hashDstEdge: Column = - concat(lit(dstPropertyGroup.name), sha2(col(dstColumnName), 256)) + concat(lit(dstPropertyGroup.name), sha2(col(dstColumnName).cast("string"), 256)) override protected[graphframes] def getData(filter: Column): DataFrame = { val filteredData = data.filter(filter) diff --git a/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala index 2f73903b0..f8b40a00a 100644 --- a/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala +++ b/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala @@ -54,11 +54,12 @@ case class VertexPropertyGroup( private[graphframes] def internalIdMapping: DataFrame = data .select(col(primaryKeyColumn).alias(EXTERNAL_ID)) - .withColumn(GraphFrame.ID, concat(lit(name), sha2(col(EXTERNAL_ID), 256))) + .withColumn(GraphFrame.ID, concat(lit(name), sha2(col(EXTERNAL_ID).cast("string"), 256))) override protected[graphframes] def getData(filter: Column): DataFrame = { val filteredData = data.filter(filter) - filteredData.select(concat(lit(name), sha2(col(primaryKeyColumn), 256)).alias(GraphFrame.ID)) + filteredData.select( + concat(lit(name), sha2(col(primaryKeyColumn).cast("string"), 256)).alias(GraphFrame.ID)) } } diff --git a/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala b/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala new file mode 100644 index 000000000..11a2a4d0d --- /dev/null +++ b/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala @@ -0,0 +1,174 @@ +package org.graphframes.propertygraph + +import org.apache.spark.sql.functions._ +import org.graphframes.GraphFrame +import org.graphframes.GraphFrameTestSparkContext +import org.graphframes.SparkFunSuite +import org.graphframes.propertygraph.property.EdgePropertyGroup +import org.graphframes.propertygraph.property.VertexPropertyGroup +import org.scalatest.BeforeAndAfterAll + +import java.security.MessageDigest + +class PropertyGraphFrameTest + extends SparkFunSuite + with GraphFrameTestSparkContext + with BeforeAndAfterAll { + var peopleMoviesGraph: PropertyGraphFrame = _ + + override def beforeAll(): Unit = { + super.beforeAll() + + // This graph represents a movie rating system with two types of vertices: 'people' (5 users: Alice, Bob, Charlie, David, Eve) + // and 'movies' (3 movies: Matrix, Inception, Interstellar). The graph has two types of edges: + // 1) 'likes' - undirected edges between people and movies with weight 1.0, representing movie preferences + // 2) 'messages' - directed edges between people with varying weights (0.3-0.9), representing communication patterns. + // The people-movie connections form a bipartite subgraph, while the messages form a directed cycle between users. + + val peopleData = spark + .createDataFrame( + Seq((1L, "Alice"), (2L, "Bob"), (3L, "Charlie"), (4L, "David"), (5L, "Eve"))) + .toDF("id", "name") + + val peopleGroup = VertexPropertyGroup("people", peopleData, "id") + + val moviesData = spark + .createDataFrame(Seq((1L, "Matrix"), (2L, "Inception"), (3L, "Interstellar"))) + .toDF("id", "title") + + val moviesGroup = VertexPropertyGroup("movies", moviesData, "id") + + val likesData = spark + .createDataFrame(Seq((1L, 1L), (1L, 2L), (2L, 1L), (3L, 2L), (4L, 3L), (5L, 2L))) + .toDF("src", "dst") + + val likesGroup = EdgePropertyGroup( + "likes", + likesData, + peopleGroup, + moviesGroup, + isDirected = false, + "src", + "dst", + lit(1.0)) + + val messagesData = spark + .createDataFrame( + Seq((1L, 2L, 5.0), (2L, 3L, 8.0), (3L, 4L, 3.0), (4L, 5L, 6.0), (5L, 1L, 9.0))) + .toDF("src", "dst", "weight") + + val messagesGroup = EdgePropertyGroup( + "messages", + messagesData, + peopleGroup, + peopleGroup, + isDirected = true, + "src", + "dst", + col("weight")) + + peopleMoviesGraph = + PropertyGraphFrame(Seq(peopleGroup, moviesGroup), Seq(likesGroup, messagesGroup)) + } + + test("projection by movies creates correct graph structure") { + val projectedGraph = peopleMoviesGraph.projectionBy("people", "movies", "likes") + + assert(projectedGraph.vertexPropertyGroups.length === 1) + assert(projectedGraph.vertexPropertyGroups.head.name === "people") + + assert(projectedGraph.edgesPropertyGroups.length === 2) + assert(projectedGraph.edgesPropertyGroups.exists(_.name === "messages")) + val projectedEdgesGroupOption = + projectedGraph.edgesPropertyGroups.find(_.name === "projected_likes") + + assert(projectedEdgesGroupOption.isDefined) + val projectedEdgesGroup = projectedEdgesGroupOption.get + + assert(projectedEdgesGroup.srcColumnName === GraphFrame.SRC) + assert(projectedEdgesGroup.dstColumnName === GraphFrame.DST) + assert(projectedEdgesGroup.weightColumnName === GraphFrame.WEIGHT) + assert(!projectedEdgesGroup.isDirected) + + val projectedEdges = projectedEdgesGroup.data + .collect() + .map(row => (row.getLong(0), row.getLong(1))) + .toSet + + // Expected edges between people who like the same movies + val expectedEdges = Set( + (1L, 2L), // Alice and Bob both like Matrix + (1L, 3L), // Alice and Charlie both like Inception + (1L, 5L), // Alice and Eve both like Inception + (3L, 5L) // Charlie and Eve both like Inception + ) + + assert(projectedEdges === expectedEdges) + } + + def sha256Hash(id: Long, groupName: String): String = { + val md = MessageDigest.getInstance("SHA-256") + val hash = md.digest(id.toString.getBytes("UTF-8")).map("%02x".format(_)).mkString + s"$groupName$hash" + } + + test("toGraphFrame with messages edges and people vertices only") { + val graph = peopleMoviesGraph.toGraphFrame( + Seq("people"), + Seq("messages"), + Map("messages" -> lit(true)), + Map("people" -> lit(true))) + + val vertices = graph.vertices.collect().map(row => row.getString(0)).toSet + val edges = graph.edges + .collect() + .map(row => (row.getString(0), row.getString(1), row.getDouble(2))) + .toSet + + // Verify vertices (all people) + val expectedVertices = Set(1L, 2L, 3L, 4L, 5L).map(sha256Hash(_, "people")) + assert(vertices === expectedVertices) + + // Verify directed message edges with weights + val expectedEdges = + Set((1L, 2L, 5.0), (2L, 3L, 8.0), (3L, 4L, 3.0), (4L, 5L, 6.0), (5L, 1L, 9.0)).map { + case (src, dst, weight) => (sha256Hash(src, "people"), sha256Hash(dst, "people"), weight) + } + assert(edges === expectedEdges) + } + + test("toGraphFrame with all groups and proper edge handling") { + val graph = peopleMoviesGraph.toGraphFrame( + Seq("people", "movies"), + Seq("messages", "likes"), + Map("messages" -> lit(true), "likes" -> lit(true)), + Map("people" -> lit(true), "movies" -> lit(true))) + + val vertices = graph.vertices.collect().toSet + val edges = graph.edges.collect().toSet + + // Verify all vertices are present + assert(vertices.size === 8) // 5 people + 3 movies + + // Verify vertex types are correctly preserved + assert(vertices.count(_.getString(0) == sha256Hash(1L, "movies")) === 1) + assert(vertices.count(_.getString(0) == sha256Hash(1L, "people")) === 1) + + // Verify edge counts and properties + val messageEdges = edges.filter(_.getDouble(2) != 1.0) + val likeEdges = edges.filter(_.getDouble(2) == 1.0) + + assert(messageEdges.size === 5) // Directed messages between people + assert(likeEdges.size === 12) // 6 original edges * 2 (undirected converted to directed) + + // Verify undirected edges were properly converted to directed pairs + val likesPairs = likeEdges.map(row => (row.getString(0), row.getString(1))).toSet + assert( + likesPairs.contains((sha256Hash(1, "people"), sha256Hash(1, "movies"))) && + likesPairs.contains((sha256Hash(1, "movies"), sha256Hash(1, "people")))) + assert( + likesPairs.contains((sha256Hash(1, "people"), sha256Hash(2, "movies"))) && + likesPairs.contains((sha256Hash(2, "movies"), sha256Hash(1, "people")))) + } + +} From ba0fc0024c65ac532e77fa899b20a3d002aa2633 Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Mon, 28 Jul 2025 11:17:06 +0200 Subject: [PATCH 9/9] from comments --- .../propertygraph/PropertyGraphFrame.scala | 23 ++++++++++++--- .../PropertyGraphFrameTest.scala | 28 +++++++++++++++++++ 2 files changed, 47 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index 956ed3565..49aa7611e 100644 --- a/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -101,13 +101,18 @@ case class PropertyGraphFrame( * Name of the vertex property group to project through * @param edgeGroup * Name of the edge property group connecting the two parts + * @param newEdgeWeight + * Optional function that takes two weight columns (Column objects) of edges as input and + * returns a new weight column. If None, a default weight of 1.0 is used for all projected + * edges. * @return * A new PropertyGraphFrame containing the projected graph */ def projectionBy( leftBiGraphPart: String, rightBiGraphPart: String, - edgeGroup: String): PropertyGraphFrame = { + edgeGroup: String, + newEdgeWeight: Option[(Column, Column) => Column] = None): PropertyGraphFrame = { require( edgeGroups(edgeGroup).srcPropertyGroup.name == leftBiGraphPart, s"Edge Property Group should have $leftBiGraphPart source group but has ${edgeGroups(edgeGroup).srcPropertyGroup.name}") @@ -116,14 +121,24 @@ case class PropertyGraphFrame( s"Edge Property Group should have $rightBiGraphPart destination group but has ${edgeGroups(edgeGroup).dstPropertyGroup.name}") val keptVPropertyGroups = vertexPropertyGroups.filterNot(g => g.name == rightBiGraphPart) val keptEPropertyGroups = edgesPropertyGroups.filterNot(g => g.name == edgeGroup) - val oldEdgesData = edgeGroups(edgeGroup).data + val oldGroup = edgeGroups(edgeGroup) + val oldEdgesData = oldGroup.data // Create new edges by joining vertices through their common neighbors val projectedEdges = oldEdgesData .as("e1") .join(oldEdgesData.as("e2"), col("e1.dst") === col("e2.dst")) .where("e1.src < e2.src") - .select(col("e1.src").alias(GraphFrame.SRC), col("e2.src").alias(GraphFrame.DST)) + .select( + col("e1.src").alias(GraphFrame.SRC), + col("e2.src").alias(GraphFrame.DST), + newEdgeWeight match { + case Some(newEdgeFunc) => + newEdgeFunc( + col(s"e1.${oldGroup.weightColumnName}"), + col(s"e2.${oldGroup.weightColumnName}")).alias(GraphFrame.WEIGHT) + case None => lit(1.0).alias(GraphFrame.WEIGHT) + }) val newEdgeGroup = EdgePropertyGroup( name = s"projected_$edgeGroup", @@ -133,7 +148,7 @@ case class PropertyGraphFrame( isDirected = false, srcColumnName = GraphFrame.SRC, dstColumnName = GraphFrame.DST, - weightColumn = lit(1.0)) + weightColumnName = GraphFrame.WEIGHT) PropertyGraphFrame(keptVPropertyGroups, keptEPropertyGroups :+ newEdgeGroup) } diff --git a/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala index d121a4541..748973b9c 100644 --- a/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala +++ b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala @@ -1,5 +1,6 @@ package org.graphframes.propertygraph +import org.apache.spark.sql.Column import org.apache.spark.sql.functions._ import org.graphframes.GraphFrame import org.graphframes.GraphFrameTestSparkContext @@ -221,6 +222,33 @@ class PropertyGraphFrameTest likesEdges.exists(e => e.getString(0) == "1" && e.getString(1) == sha256Hash(1L, "people"))) } + test("projection with custom weight function") { + val projectedGraph = peopleMoviesGraph.projectionBy( + "people", + "movies", + "likes", + Some((leftWeight: Column, rightWeight: Column) => leftWeight + rightWeight)) + + val projectedEdgesGroupOption = + projectedGraph.edgesPropertyGroups.find(_.name === "projected_likes") + assert(projectedEdgesGroupOption.isDefined) + + val projectedEdges = projectedEdgesGroupOption.get.data + .collect() + .map(row => (row.getLong(0), row.getLong(1), row.getDouble(2))) + .toSet + + // Expected edges between people who like the same movies with sum of their weights + val expectedEdges = Set( + (1L, 2L, 2.0), // Alice and Bob both like Matrix (1.0 + 1.0) + (1L, 3L, 2.0), // Alice and Charlie both like Inception (1.0 + 1.0) + (1L, 5L, 2.0), // Alice and Eve both like Inception (1.0 + 1.0) + (3L, 5L, 2.0) // Charlie and Eve both like Inception (1.0 + 1.0) + ) + + assert(projectedEdges === expectedEdges) + } + test("joinVertices withConnectedComponents") { // Convert to GraphFrame with all vertices and edges val graph = peopleMoviesGraph.toGraphFrame(