graphframes · SemyonSinchenko · Oct 28, 2025 · Oct 24, 2025 · Oct 24, 2025 · Oct 28, 2025
diff --git a/connect/src/main/protobuf/graphframes.proto b/connect/src/main/protobuf/graphframes.proto
@@ -35,6 +35,7 @@ message GraphFramesAPI {
     SVDPlusPlus svd_plus_plus = 18;
     TriangleCount triangle_count = 19;
     Triplets triplets = 20;
+    MaximalIndependentSet mis = 22;
     KCore kcore = 21;
   }
 }
@@ -189,6 +190,13 @@ message TriangleCount {
 
 message Triplets {}
 
+message MaximalIndependentSet {
+  int32 checkpoint_interval = 1;
+  optional StorageLevel storage_level = 2;
+  bool use_local_checkpoints = 3;
+  int64 seed = 4;
+}
+
 message KCore {
   bool use_local_checkpoints = 1;
   int32 checkpoint_interval = 2;

diff --git a/connect/src/main/scala/org/apache/spark/sql/graphframes/GraphFramesConnectUtils.scala b/connect/src/main/scala/org/apache/spark/sql/graphframes/GraphFramesConnectUtils.scala
@@ -406,6 +406,19 @@ object GraphFramesConnectUtils {
       case proto.GraphFramesAPI.MethodCase.TRIPLETS => {
         graphFrame.triplets
       }
+      case proto.GraphFramesAPI.MethodCase.MIS => {
+        val mis = graphFrame.maximalIndependentSet
+          .setCheckpointInterval(apiMessage.getMis.getCheckpointInterval)
+          .setUseLocalCheckpoints(apiMessage.getMis.getUseLocalCheckpoints)
+
+        if (apiMessage.getMis.hasStorageLevel) {
+          mis
+            .setIntermediateStorageLevel(parseStorageLevel(apiMessage.getMis.getStorageLevel))
+            .run(apiMessage.getMis.getSeed)
+        } else {
+          mis.run(apiMessage.getMis.getSeed)
+        }
+      }
       case proto.GraphFramesAPI.MethodCase.KCORE => {
         var kCoreBuilder =
           graphFrame.kCore

diff --git a/core/src/main/scala/org/graphframes/GraphFrame.scala b/core/src/main/scala/org/graphframes/GraphFrame.scala
@@ -797,6 +797,15 @@ class GraphFrame private (
    */
   def detectingCycles: DetectingCycles = new DetectingCycles(this)
 
+  /**
+   * Maximal Independent Set algorithm.
+   *
+   * See [[org.graphframes.lib.MaximalIndependentSet]] for more details.
+   *
+   * @group stdlib
+   */
+  def maximalIndependentSet: MaximalIndependentSet = new MaximalIndependentSet(this)
+
   /**
    * Converts the directed graph into an undirected graph by ensuring that all directed edges are
    * bidirectional. For every directed edge (src, dst), a corresponding edge (dst, src) is added.

diff --git a/core/src/main/scala/org/graphframes/lib/MaximalIndependentSet.scala b/core/src/main/scala/org/graphframes/lib/MaximalIndependentSet.scala
@@ -0,0 +1,225 @@
+package org.graphframes.lib
+
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.functions.*
+import org.apache.spark.sql.types.DoubleType
+import org.apache.spark.storage.StorageLevel
+import org.graphframes.GraphFrame
+import org.graphframes.Logging
+import org.graphframes.WithCheckpointInterval
+import org.graphframes.WithIntermediateStorageLevel
+import org.graphframes.WithLocalCheckpoints
+
+import java.io.IOException
+
+/**
+ * This class implements a distributed algorithm for finding a Maximal Independent Set (MIS) in a
+ * graph.
+ *
+ * An MIS is a set of vertices such that no two vertices in the set are adjacent (i.e., there is
+ * no edge between any two vertices in the set), and the set is maximal, meaning that adding any
+ * other vertex to the set would violate the independence property. Note that this implementation
+ * finds a maximal (but not necessarily maximum) independent set; that is, it ensures no more
+ * vertices can be added to the set, but does not guarantee that the set has the largest possible
+ * number of vertices among all possible independent sets in the graph.
+ *
+ * The algorithm implemented here is based on the paper: Ghaffari, Mohsen. "An improved
+ * distributed algorithm for maximal independent set." Proceedings of the twenty-seventh annual
+ * ACM-SIAM symposium on Discrete algorithms. Society for Industrial and Applied Mathematics,
+ * 2016.
+ *
+ * Note: This is a randomized, non-deterministic algorithm. The result may vary between runs even
+ * if a fixed random seed is provided because how Apache Spark works.
+ *
+ * @param graph
+ */
+class MaximalIndependentSet private[graphframes] (private val graph: GraphFrame)
+    extends Serializable
+    with WithIntermediateStorageLevel
+    with WithCheckpointInterval
+    with WithLocalCheckpoints {
+  def run(seed: Long): DataFrame = {
+    MaximalIndependentSet.run(
+      graph,
+      checkpointInterval,
+      useLocalCheckpoints,
+      intermediateStorageLevel,
+      seed)
+  }
+}
+
+object MaximalIndependentSet extends Serializable with Logging {
+  private val probCol = "prob"
+  private val degCol = "effectiveDegree"
+  private val isNominated = "isNominated"
+  private val notJoinedMISCol = "notJoinMIS"
+  private val isMIS = "isMIS"
+
+  private def run(
+      graph: GraphFrame,
+      checkpointInterval: Int,
+      useLocalCheckpoints: Boolean,
+      storageLevel: StorageLevel,
+      seed: Long): DataFrame = {
+    // initial p = 1/2
+    var vertices =
+      graph.vertices
+        .select(col(GraphFrame.ID), lit(0.5).cast(DoubleType).alias(probCol))
+        .persist(storageLevel)
+
+    // make edges undirected and de-duplicate
+    // persist() for future usage
+    val edges = graph.edges
+      .select(GraphFrame.SRC, GraphFrame.DST)
+      .union(
+        graph.edges.select(
+          col(GraphFrame.DST).alias(GraphFrame.SRC),
+          col(GraphFrame.SRC).alias(GraphFrame.DST)))
+      .filter(col(GraphFrame.SRC) =!= col(GraphFrame.DST))
+      .distinct()
+      .persist(storageLevel)
+
+    var misDF = graph.vertices.select(col(GraphFrame.ID), lit(false).alias(isMIS))
+
+    var i = 0
+    var converged = false
+    val spark = graph.vertices.sparkSession
+
+    val shouldCheckpoint = checkpointInterval > 0
+    if (!useLocalCheckpoints && spark.sparkContext.getCheckpointDir.isEmpty) {
+      // Spark-Connect workaround
+      spark.sparkContext
+        .setCheckpointDir(spark.conf
+          .getOption("spark.checkpoint.dir") match {
+          case Some(d) => d
+          case None =>
+            throw new IOException(
+              "Checkpoint directory is not set. Please set it first using sc.setCheckpointDir()" +
+                "or by specifying the conf 'spark.checkpoint.dir'.")
+        })
+    }
+
+    val rng = new util.Random(seed)
+
+    // randomized algorithms are not working with AQE well
+    val originalAQE = spark.conf.get("spark.sql.adaptive.enabled")
+    try {
+      spark.conf.set("spark.sql.adaptive.enabled", "false")
+
+      while (!converged) {
+        val iterSeed = rng.nextLong()
+        // compute effective degree as a sum of nbrs p
+        val effectiveDegrees =
+          edges
+            .join(vertices, col(GraphFrame.ID) === col(GraphFrame.DST))
+            .groupBy(GraphFrame.SRC)
+            .agg(sum(col(probCol)).alias(degCol))
+
+        // update p per vertex by condition:
+        // if effective degree >= 2 then p / 2
+        // else min(2p, 1/2)
+        //
+        // + mark vertices based on p
+        val probs = vertices
+          .join(effectiveDegrees, col(GraphFrame.ID) === col(GraphFrame.SRC))
+          .drop(GraphFrame.SRC)
+          .withColumn(
+            probCol,
+            when(col(degCol) >= lit(2), col(probCol) / lit(2.0)).otherwise(
+              when(lit(2) * col(probCol) <= lit(0.5), lit(2) * col(probCol)).otherwise(lit(0.5))))
+          .withColumn(isNominated, col(probCol) >= rand(iterSeed))
+          .select(GraphFrame.ID, isNominated, probCol)
+          .persist(storageLevel)
+
+        val isolatedVertices =
+          vertices
+            .join(probs.select(col(GraphFrame.ID)), Seq(GraphFrame.ID), "left_anti")
+            .select(GraphFrame.ID)
+
+        // if no nbr of v is marked and v is marked,
+        // v is joined MIS and removed with all it's nbrs
+        val isJoinedMIS = probs
+          .join(
+            edges
+              .join(probs, col(GraphFrame.ID) === col(GraphFrame.DST))
+              .groupBy(GraphFrame.SRC)
+              .agg(bool_or(col(isNominated)).alias(notJoinedMISCol)),
+            col(GraphFrame.SRC) === col(GraphFrame.ID))
+          .select(GraphFrame.ID, probCol, isNominated, notJoinedMISCol)
+
+        val joinedMIS =
+          isJoinedMIS.filter((!col(notJoinedMISCol)) && col(isNominated)).select(GraphFrame.ID)
+
+        // update curent MIS
+        val updatedMIS = misDF
+          .join(
+            isolatedVertices.select(col(GraphFrame.ID), lit(true).alias("f")),
+            Seq(GraphFrame.ID),
+            "left")
+          .select(col(GraphFrame.ID), (col(isMIS) || col("f")).alias(isMIS))
+          .join(
+            joinedMIS.select(col(GraphFrame.ID), lit(true).alias("f")),
+            Seq(GraphFrame.ID),
+            "left")
+          .select(col(GraphFrame.ID), (col(isMIS) || col("f")).alias(isMIS))
+          .persist(storageLevel)
+
+        // We cannot not checkpoint current MIS, otherwise it is almost not working.
+        if (useLocalCheckpoints) {
+          val newMis = updatedMIS.localCheckpoint(eager = true)
+          newMis.count()
+          misDF.unpersist()
+          misDF = newMis
+        } else {
+          val newMis = updatedMIS.checkpoint(eager = true)
+          newMis.count()
+          misDF.unpersist()
+          misDF = newMis
+        }
+
+        val neighborsOfMIS = edges
+          .join(joinedMIS, col(GraphFrame.ID) === col(GraphFrame.DST))
+          .select(col(GraphFrame.SRC))
+
+        val updatedVertices = probs
+          .join(joinedMIS, Seq(GraphFrame.ID), "left_anti")
+          .join(neighborsOfMIS, col(GraphFrame.ID) === col(GraphFrame.SRC), "left_anti")
+          .select(GraphFrame.ID, probCol)
+
+        // checkpointing of vertices
+        if (shouldCheckpoint && (i % checkpointInterval == 0)) {
+          if (useLocalCheckpoints) {
+            vertices = updatedVertices.localCheckpoint(eager = true)
+          } else {
+            vertices = updatedVertices.checkpoint(eager = true)
+          }
+        } else {
+          vertices = updatedVertices
+        }
+
+        // algorithm stops if no more vertex left
+        converged = vertices.isEmpty
+
+        updatedVertices.unpersist()
+        probs.unpersist()
+
+        logInfo(s"iteration $i finished, vertices left: ${vertices.count()}")
+        i += 1
+      }
+
+      vertices.unpersist(true)
+      edges.unpersist(true)
+
+      val mis = misDF.filter(col(isMIS)).select(GraphFrame.ID).persist(storageLevel)
+      // materialize
+      mis.count()
+      resultIsPersistent()
+      misDF.unpersist(true)
+
+      mis
+    } finally {
+      // Restore original AQE setting
+      spark.conf.set("spark.sql.adaptive.enabled", originalAQE)
+    }
+  }
+}