Skip to content

chore: add useLabelsAsComponents configuration for connected components #623

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SemyonSinchenko merged 2 commits into from
Jul 3, 2025
Merged

chore: add useLabelsAsComponents configuration for connected components #623

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ce03bf3
Add `useLabelsAsComponents` configuration for connected components
SemyonSinchenko Jul 3, 2025
a521af3
updates from comments
SemyonSinchenko Jul 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions src/main/scala/org/apache/spark/sql/graphframes/GraphFramesConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,16 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.storage.StorageLevel

object GraphFramesConf {
private val USE_LABELS_AS_COMPONENTS =
SQLConf
.buildConf("spark.graphframes.useLabelsAsComponents")
.doc(""" Tells the connected components algorithm to use (default: "true") labels as components in the output
| DataFrame. If set to "false", randomly generated labels with the data type LONG will returned.
|""".stripMargin)
.version("0.9.0")
.booleanConf
.createWithDefault(true)

private val CONNECTED_COMPONENTS_ALGORITHM =
SQLConf
.buildConf("spark.graphframes.connectedComponents.algorithm")
Expand Down Expand Up @@ -93,4 +103,6 @@ object GraphFramesConf {
case _ => None
}
}

def getUseLabelsAsComponents: Boolean = get(USE_LABELS_AS_COMPONENTS).get.toBoolean
}
2 changes: 1 addition & 1 deletion src/main/scala/org/graphframes/lib/ConnectedComponents.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -340,7 +340,7 @@ object ConnectedComponents extends Logging {
vv(ATTR),
when(ee(SRC).isNull, vv(ID)).otherwise(ee(SRC)).as(COMPONENT),
col(ATTR + "." + ID).as(ID))
val output = if (graph.hasIntegralIdType) {
val output = if (graph.hasIntegralIdType || !GraphFramesConf.getUseLabelsAsComponents) {
indexedLabel
.select(col(s"$ATTR.*"), col(COMPONENT))
.persist(intermediateStorageLevel)
Expand Down
20 changes: 20 additions & 0 deletions src/test/scala/org/graphframes/lib/ConnectedComponentsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.types.DataTypes
import org.apache.spark.sql.types.LongType
import org.apache.spark.storage.StorageLevel
import org.graphframes.GraphFrame._
import org.graphframes._
Expand Down Expand Up @@ -75,6 +76,25 @@ class ConnectedComponentsSuite extends SparkFunSuite with GraphFrameTestSparkCon
assertComponents(components, expected)
}

test("using labels as components") {
val vertices = spark.createDataFrame(Seq("a", "b", "c", "d", "e").map(Tuple1.apply)).toDF(ID)
val edges = spark.createDataFrame(Seq.empty[(String, String)]).toDF(SRC, DST)
val g = GraphFrame(vertices, edges)
val components = g.connectedComponents.run()
val expected = Seq("a", "b", "c", "d", "e").map(Set(_)).toSet
assertComponents(components, expected)
}

test("don't using labels as components") {
spark.conf.set("spark.graphframes.useLabelsAsComponents", "false")
val vertices = spark.createDataFrame(Seq("a", "b", "c", "d", "e").map(Tuple1.apply)).toDF(ID)
val edges = spark.createDataFrame(Seq.empty[(String, String)]).toDF(SRC, DST)
val g = GraphFrame(vertices, edges)
val components = g.connectedComponents.run()
assert(components.schema("component").dataType == LongType)
spark.conf.set("spark.graphframes.useLabelsAsComponents", "true")
}

test("two connected vertices") {
val v = spark.createDataFrame(List((0L, "a0", "b0"), (1L, "a1", "b1"))).toDF("id", "A", "B")
val e = spark.createDataFrame(List((0L, 1L, "a01", "b01"))).toDF("src", "dst", "A", "B")
Expand Down