Skip to content

feat: Randomized Contraction CC #776

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 3 commits into from
Jan 26, 2026
Merged

feat: Randomized Contraction CC #776

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3e98e89
implementation of new CC algorithm
SemyonSinchenko Jan 22, 2026
09d9a06
Merge remote-tracking branch 'graphframes/main' into 729-randomized-c…
SemyonSinchenko Jan 26, 2026
ab2e68c
from comments
SemyonSinchenko Jan 26, 2026
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
Prev Previous commit
from comments
  • Loading branch information
@SemyonSinchenko
SemyonSinchenko committed Jan 26, 2026
commit ab2e68cc75100fb14a5febaefa753af3675c3ca6
12 changes: 8 additions & 4 deletions ...aphframes/lib/ConnectedComponentsV2.scala → ...aphframes/lib/RandomizedContraction.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,12 @@ import org.apache.spark.sql.functions.*
import org.apache.spark.sql.graphframes.expressions.FiniteAXPlusB
import org.apache.spark.storage.StorageLevel
import org.graphframes.GraphFrame
import org.graphframes.GraphFrame.DST
import org.graphframes.GraphFrame.ID
import org.graphframes.GraphFrame.LONG_DST
import org.graphframes.GraphFrame.LONG_ID
import org.graphframes.GraphFrame.LONG_SRC
import org.graphframes.GraphFrame.SRC
import org.graphframes.Logging

import java.io.IOException
Expand All @@ -24,11 +30,9 @@ import scala.util.Random
* The algorithm contracts the graph iteratively using random linear functions, until no edges
* remain, then reconstructs the component identifiers.
*/
private[graphframes] object ConnectedComponentsV2 extends Logging with Serializable {
private[graphframes] object RandomizedContraction extends Logging with Serializable {
private val CHECKPOINT_NAME_PREFIX = "randomized-contraction"

import GraphFrame.{ID, SRC, DST, LONG_SRC, LONG_DST, LONG_ID}

private def prepare(graph: GraphFrame): GraphFrame = {
val vertices = graph.indexedVertices
.select(col(LONG_ID).as(ID))
Expand Down Expand Up @@ -228,7 +232,7 @@ private[graphframes] object ConnectedComponentsV2 extends Logging with Serializa
}

outputComponents.persist(intermediateStorageLevel)
// meterialize to be able to clean up everything
// materialize to be able to clean up everything
outputComponents.count()

// clean-up
Expand Down
52 changes: 26 additions & 26 deletions core/src/test/scala/org/graphframes/lib/RandomizedContractionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,9 @@ import org.graphframes.examples.Graphs

class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkContext {

test("ConnectedComponentsV2: empty graph") {
test("RandomizedContraction: empty graph") {
val graph = Graphs.empty[Long]
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -23,12 +23,12 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: single isolated vertex") {
test("RandomizedContraction: single isolated vertex") {
val vertices = spark.createDataFrame(List((0L, "a", "b"))).toDF("id", "vattr", "gender")
val e =
spark.createDataFrame(List((0L, 0L, 1L))).toDF("src", "dst", "test").filter("src > 10")
val graph = GraphFrame(vertices, e)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -38,12 +38,12 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: two connected vertices") {
test("RandomizedContraction: two connected vertices") {
val vertices =
spark.createDataFrame(List((0L, "a0", "b0"), (1L, "a1", "b1"))).toDF("id", "A", "B")
val edges = spark.createDataFrame(List((0L, 1L, "a01", "b01"))).toDF("src", "dst", "A", "B")
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -55,10 +55,10 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: chain graph") {
test("RandomizedContraction: chain graph") {
val n = 5L
val graph = Graphs.chain(n)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -68,13 +68,13 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: disconnected vertices") {
test("RandomizedContraction: disconnected vertices") {
val n = 5L
val vertices = spark.range(n).toDF(GraphFrame.ID)
val edges =
spark.createDataFrame(Seq.empty[(Long, Long)]).toDF(GraphFrame.SRC, GraphFrame.DST)
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -84,13 +84,13 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: two separate components") {
test("RandomizedContraction: two separate components") {
val vertices = spark.range(6L).toDF(GraphFrame.ID)
val edges = spark
.createDataFrame(Seq((0L, 1L), (1L, 2L), (2L, 0L), (3L, 4L), (4L, 5L), (5L, 3L)))
.toDF(GraphFrame.SRC, GraphFrame.DST)
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -102,13 +102,13 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: with dangling vertices") {
test("RandomizedContraction: with dangling vertices") {
val vertices = spark.range(8L).toDF(GraphFrame.ID)
val edges = spark
.createDataFrame(Seq((0L, 1L), (1L, 2L), (2L, 0L), (3L, 4L), (4L, 5L), (5L, 3L)))
.toDF(GraphFrame.SRC, GraphFrame.DST)
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -120,13 +120,13 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: useLabelsAsComponents with string IDs") {
test("RandomizedContraction: useLabelsAsComponents with string IDs") {
val vertices =
spark.createDataFrame(Seq("a", "b", "c", "d").map(Tuple1.apply)).toDF(GraphFrame.ID)
val edges =
spark.createDataFrame(Seq(("a", "b"), ("b", "c"))).toDF(GraphFrame.SRC, GraphFrame.DST)
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = true,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -138,13 +138,13 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: useLabelsAsComponents with long IDs") {
test("RandomizedContraction: useLabelsAsComponents with long IDs") {
val vertices =
spark.createDataFrame(Seq(1L, 2L, 3L, 4L).map(Tuple1.apply)).toDF(GraphFrame.ID)
val edges =
spark.createDataFrame(Seq((1L, 2L), (2L, 3L))).toDF(GraphFrame.SRC, GraphFrame.DST)
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = true,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -156,11 +156,11 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: no parquet file leaks") {
test("RandomizedContraction: no parquet file leaks") {
val graph = Graphs.chain(3L)
val initialParquetFiles = listParquetFiles()

val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -172,11 +172,11 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: no memory leaks") {
test("RandomizedContraction: no memory leaks") {
val priorCachedCount = spark.sparkContext.getPersistentRDDs.size

val graph = Graphs.chain(10L)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -189,15 +189,15 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: large long IDs") {
test("RandomizedContraction: large long IDs") {
val max = Long.MaxValue
val chain = Graphs.chain(10L)
val vertices = chain.vertices.select((col(GraphFrame.ID) - lit(max)).as(GraphFrame.ID))
val edges = chain.edges.select(
(col(GraphFrame.SRC) - lit(max)).as(GraphFrame.SRC),
(col(GraphFrame.DST) - lit(max)).as(GraphFrame.DST))
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand All @@ -207,13 +207,13 @@ class RandomizedContractionSuite extends SparkFunSuite with GraphFrameTestSparkC
assertFunctionRegistryClean()
}

test("ConnectedComponentsV2: directed edges still produce connected components") {
test("RandomizedContraction: directed edges still produce connected components") {
val vertices = spark.range(5L).toDF(GraphFrame.ID)
val edges = spark
.createDataFrame(Seq((0L, 4L), (4L, 3L), (2L, 3L), (2L, 1L)))
.toDF(GraphFrame.SRC, GraphFrame.DST)
val graph = GraphFrame(vertices, edges)
val components = ConnectedComponentsV2.run(
val components = RandomizedContraction.run(
inputGraph = graph,
useLabelsAsComponents = false,
intermediateStorageLevel = StorageLevel.MEMORY_AND_DISK,
Expand Down