graphframes · james-willis · Sep 2, 2026 · Sep 2, 2026 · Sep 2, 2026
diff --git a/core/src/main/scala/org/graphframes/lib/DetectingCycles.scala b/core/src/main/scala/org/graphframes/lib/DetectingCycles.scala
@@ -85,6 +85,7 @@ object DetectingCycles {
       x => array_append(x, col(GraphFrame.ID)))
 
     preparedGraph.pregel
+      .setJobDescriptionPrefix("GraphFrames DetectingCycles")
       .setCheckpointInterval(checkpointInterval)
       .setUseLocalCheckpoints(useLocalCheckpoints)
       .setIntermediateStorageLevel(intermediateStorageLevel)

diff --git a/core/src/main/scala/org/graphframes/lib/KCore.scala b/core/src/main/scala/org/graphframes/lib/KCore.scala
@@ -78,6 +78,7 @@ object KCore extends Serializable with Logging {
 
     try {
       val pregel = preparedGraph.pregel
+        .setJobDescriptionPrefix("GraphFrames KCore")
         .setMaxIter(Int.MaxValue)
         .setIntermediateStorageLevel(storageLevel)
         .setCheckpointInterval(checkpointInterval)

diff --git a/core/src/main/scala/org/graphframes/lib/LabelPropagation.scala b/core/src/main/scala/org/graphframes/lib/LabelPropagation.scala
@@ -110,6 +110,7 @@ private object LabelPropagation {
       graph.edges.select(GraphFrame.SRC, GraphFrame.DST))
 
     var pregel = preparedGraph.pregel
+      .setJobDescriptionPrefix("GraphFrames LabelPropagation")
       .withVertexColumn(LABEL_ID, col(GraphFrame.ID).alias(LABEL_ID), keyWithMaxValue(Pregel.msg))
       .setMaxIter(maxIter)
       .setStopIfAllNonActiveVertices(true)

diff --git a/core/src/main/scala/org/graphframes/lib/MaximalIndependentSet.scala b/core/src/main/scala/org/graphframes/lib/MaximalIndependentSet.scala
@@ -5,6 +5,7 @@ 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.JobDescription
 import org.graphframes.Logging
 import org.graphframes.WithCheckpointInterval
 import org.graphframes.WithIntermediateStorageLevel
@@ -103,10 +104,13 @@ object MaximalIndependentSet extends Serializable with Logging {
 
     // randomized algorithms are not working with AQE well
     val originalAQE = spark.conf.get("spark.sql.adaptive.enabled")
+    val previousJobDescription =
+      spark.sparkContext.getLocalProperty(JobDescription.JOB_DESCRIPTION_KEY)
     try {
       spark.conf.set("spark.sql.adaptive.enabled", "false")
 
       while (!converged) {
+        spark.sparkContext.setJobDescription(s"GraphFrames MaximalIndependentSet: iteration $i")
         val iterSeed = rng.nextLong()
         // compute effective degree as a sum of nbrs p
         val effectiveDegrees =
@@ -210,6 +214,8 @@ object MaximalIndependentSet extends Serializable with Logging {
       vertices.unpersist(true)
       edges.unpersist(true)
 
+      spark.sparkContext.setJobDescription(
+        "GraphFrames MaximalIndependentSet: materializing final result")
       val mis = misDF.filter(col(isMIS)).select(GraphFrame.ID).persist(storageLevel)
       // materialize
       mis.count()
@@ -220,6 +226,8 @@ object MaximalIndependentSet extends Serializable with Logging {
     } finally {
       // Restore original AQE setting
       spark.conf.set("spark.sql.adaptive.enabled", originalAQE)
+      spark.sparkContext
+        .setLocalProperty(JobDescription.JOB_DESCRIPTION_KEY, previousJobDescription)
     }
   }
 }
diff --git a/core/src/main/scala/org/graphframes/lib/Pregel.scala b/core/src/main/scala/org/graphframes/lib/Pregel.scala
@@ -27,8 +27,10 @@ import org.apache.spark.sql.functions.struct
 import org.apache.spark.sql.graphframes.SparkShims
 import org.graphframes.GraphFrame
 import org.graphframes.GraphFrame.*
+import org.graphframes.JobDescription
 import org.graphframes.Logging
 import org.graphframes.WithIntermediateStorageLevel
+import org.graphframes.WithJobDescriptionPrefix
 import org.graphframes.WithLocalCheckpoints
 
 import java.io.IOException
@@ -93,7 +95,10 @@ import scala.util.control.Breaks.breakable
 class Pregel(val graph: GraphFrame)
     extends Logging
     with WithLocalCheckpoints
-    with WithIntermediateStorageLevel {
+    with WithIntermediateStorageLevel
+    with WithJobDescriptionPrefix {
+
+  override protected def defaultJobDescriptionPrefix: String = "GraphFrames Pregel"
 
   private val withVertexColumnList = collection.mutable.ListBuffer.empty[(String, Column, Column)]
 
@@ -410,6 +415,12 @@ class Pregel(val graph: GraphFrame)
       withVertexColumnList.size > 0,
       "There should be at least one additional vertex columns for updating.")
 
+    JobDescription.withRestoredJobDescription(graph.spark) {
+      runAlgorithm()
+    }
+  }
+
+  private def runAlgorithm(): DataFrame = {
     val sendMsgsColList = sendMsgs.toList.map { case (id, msg) =>
       struct(id.as(ID), msg.as("msg"))
     }
@@ -466,6 +477,9 @@ class Pregel(val graph: GraphFrame)
 
     var iteration = 1
 
+    // Avoid "iteration 3 / 2147483647" for algorithms that rely on early stopping.
+    val maxIterSuffix = if (maxIter == Int.MaxValue) "" else s" / $maxIter"
+
     val shouldCheckpoint = checkpointInterval > 0
 
     if (shouldCheckpoint && graph.spark.sparkContext.getCheckpointDir.isEmpty && !useLocalCheckpoints) {
@@ -490,6 +504,8 @@ class Pregel(val graph: GraphFrame)
     breakable {
       while (iteration <= maxIter) {
         logInfo(s"start Pregel iteration $iteration / $maxIter")
+        graph.spark.sparkContext.setJobDescription(
+          s"$getJobDescriptionPrefix: iteration $iteration$maxIterSuffix")
         val currRoundPersistent = scala.collection.mutable.Queue[DataFrame]()
         currRoundPersistent.enqueue(currentVertices.persist(intermediateStorageLevel))
 
@@ -588,6 +604,8 @@ class Pregel(val graph: GraphFrame)
       }
     }
 
+    graph.spark.sparkContext.setJobDescription(
+      s"$getJobDescriptionPrefix: materializing final result")
     val res = currentVertices.persist(intermediateStorageLevel)
     res.count()
     while (lastRoundPersistent.nonEmpty) {

diff --git a/core/src/main/scala/org/graphframes/lib/RandomizedContraction.scala b/core/src/main/scala/org/graphframes/lib/RandomizedContraction.scala
@@ -15,6 +15,7 @@ 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.JobDescription
 import org.graphframes.Logging
 
 import java.io.IOException
@@ -57,6 +58,8 @@ private[graphframes] object RandomizedContraction extends Logging with Serializa
     val sc = spark.sparkContext
     val runId = UUID.randomUUID().toString.takeRight(8)
     val logPrefix = s"[CC $runId]"
+    val jobDescriptionPrefix = s"GraphFrames ConnectedComponents [$runId]"
+    val previousJobDescription = sc.getLocalProperty(JobDescription.JOB_DESCRIPTION_KEY)
 
     val checkpointDir = sc.getCheckpointDir
       .map { d =>
@@ -100,6 +103,7 @@ private[graphframes] object RandomizedContraction extends Logging with Serializa
     def axpb(a: Long, x: Column, b: Long): Column = call_function("_axpb", lit(a), x, lit(b))
 
     try {
+      sc.setJobDescription(s"$jobDescriptionPrefix: preparing graph")
       var rA = 0L
       var graphSize = edges.count()
       var ccRepresentatives: DataFrame = null
@@ -117,6 +121,7 @@ private[graphframes] object RandomizedContraction extends Logging with Serializa
 
       while (graphSize > 0) {
         logInfo(s"iteration ${iter}, edges left ${graphSize}")
+        sc.setJobDescription(s"$jobDescriptionPrefix: iteration $iter, $graphSize edges left")
         iter += 1
         rA = 0L
         while (rA == 0L) {
@@ -177,6 +182,7 @@ private[graphframes] object RandomizedContraction extends Logging with Serializa
 
       while (iter > 1) {
         iter -= 1
+        sc.setJobDescription(s"$jobDescriptionPrefix: reverse transformation step $iter")
         val poppedA = stackA.pop()
         val poppedB = stackB.pop()
 
@@ -248,6 +254,7 @@ private[graphframes] object RandomizedContraction extends Logging with Serializa
               .alias(ConnectedComponents.COMPONENT))
       }
 
+      sc.setJobDescription(s"$jobDescriptionPrefix: materializing final result")
       outputComponents.persist(intermediateStorageLevel)
       // materialize to be able to clean up everything
       outputComponents.count()
@@ -261,6 +268,7 @@ private[graphframes] object RandomizedContraction extends Logging with Serializa
 
       outputComponents
     } finally {
+      sc.setLocalProperty(JobDescription.JOB_DESCRIPTION_KEY, previousJobDescription)
       // to be 100% sure;
       edges.unpersist()
       val dereg = functionRegistry.dropFunction(FunctionIdentifier("_axpb"))

diff --git a/core/src/main/scala/org/graphframes/lib/ShortestPaths.scala b/core/src/main/scala/org/graphframes/lib/ShortestPaths.scala
@@ -215,6 +215,7 @@ private object ShortestPaths extends Logging {
     // 2. If new message can improve distances send it
     // 3. Collect and aggregate messages
     val pregel = preparedGraph.pregel
+      .setJobDescriptionPrefix("GraphFrames ShortestPaths")
       .setIntermediateStorageLevel(intermediateStorageLevel)
       .setMaxIter(Int.MaxValue) // That is how the GraphX implementation works
       .withVertexColumn(

diff --git a/core/src/main/scala/org/graphframes/lib/StructureAwareLabelPropagation.scala b/core/src/main/scala/org/graphframes/lib/StructureAwareLabelPropagation.scala
@@ -181,6 +181,7 @@ class StructureAwareLabelPropagation private[graphframes] (private val graph: Gr
     val pregel = preparedGraph.pregel
 
     pregel
+      .setJobDescriptionPrefix("GraphFrames StructureAwareLabelPropagation")
       .setMaxIter(maxIterChecked)
       .setCheckpointInterval(checkpointInterval)
       .setUseLocalCheckpoints(useLocalCheckpoints)

diff --git a/core/src/main/scala/org/graphframes/lib/TwoPhase.scala b/core/src/main/scala/org/graphframes/lib/TwoPhase.scala
@@ -31,6 +31,7 @@ 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.JobDescription
 import org.graphframes.Logging
 
 import java.io.IOException
@@ -259,13 +260,16 @@ private[graphframes] object TwoPhase extends Logging {
     val spark = graph.spark
     val sc = spark.sparkContext
     val originalAQE = spark.conf.get("spark.sql.adaptive.enabled")
+    val previousJobDescription = sc.getLocalProperty(JobDescription.JOB_DESCRIPTION_KEY)
 
     try {
       spark.conf.set("spark.sql.adaptive.enabled", "false")
 
       val runId = UUID.randomUUID().toString.takeRight(8)
       val logPrefix = s"[CC $runId]"
       logInfo(s"$logPrefix Start connected components with run ID $runId.")
+      val jobDescriptionPrefix = s"GraphFrames ConnectedComponents [$runId]"
+      sc.setJobDescription(s"$jobDescriptionPrefix: preparing graph")
 
       val shouldCheckpoint = checkpointInterval > 0
       val checkpointDir: Option[String] = if (useLocalCheckpoints) { None }
@@ -314,6 +318,7 @@ private[graphframes] object TwoPhase extends Logging {
 
       var lastRoundPersistedDFs = Seq[DataFrame](ee, minNbrs1)
       while (!converged) {
+        sc.setJobDescription(s"$jobDescriptionPrefix: iteration $iteration")
         var currRoundPersistedDFs = Seq[DataFrame]()
 
         // large-star step
@@ -430,6 +435,7 @@ private[graphframes] object TwoPhase extends Logging {
       logInfo(s"$logPrefix Connected components converged in ${iteration - 1} iterations.")
       logInfo(s"$logPrefix Join and return component assignments with original vertex IDs.")
 
+      sc.setJobDescription(s"$jobDescriptionPrefix: materializing final result")
       val output = buildOutput(graph, vv, ee, useLabelsAsComponents)
         .persist(intermediateStorageLevel)
 
@@ -447,6 +453,7 @@ private[graphframes] object TwoPhase extends Logging {
       output
     } finally {
       spark.conf.set("spark.sql.adaptive.enabled", originalAQE)
+      sc.setLocalProperty(JobDescription.JOB_DESCRIPTION_KEY, previousJobDescription)
     }
   }
 
@@ -464,11 +471,37 @@ private[graphframes] object TwoPhase extends Logging {
       isGraphPrepared: Boolean,
       optStartIter: Int = 2,
       sparsityThreshold: Double = 2.0,
-      shrinkageThreshold: Double = 2.0): DataFrame = {
+      shrinkageThreshold: Double = 2.0): DataFrame =
+    JobDescription.withRestoredJobDescription(graph.spark) {
+      runAQEInternal(
+        graph,
+        checkpointInterval,
+        intermediateStorageLevel,
+        useLabelsAsComponents,
+        useLocalCheckpoints,
+        isGraphPrepared,
+        optStartIter,
+        sparsityThreshold,
+        shrinkageThreshold)
+    }
+
+  private def runAQEInternal(
+      graph: GraphFrame,
+      checkpointInterval: Int,
+      intermediateStorageLevel: StorageLevel,
+      useLabelsAsComponents: Boolean,
+      useLocalCheckpoints: Boolean,
+      isGraphPrepared: Boolean,
+      optStartIter: Int,
+      sparsityThreshold: Double,
+      shrinkageThreshold: Double): DataFrame = {
 
+    val sc = graph.spark.sparkContext
     val runId = UUID.randomUUID().toString.takeRight(8)
     val logPrefix = s"[CC $runId]"
     logInfo(s"$logPrefix Start connected components with run ID $runId.")
+    val jobDescriptionPrefix = s"GraphFrames ConnectedComponents [$runId]"
+    sc.setJobDescription(s"$jobDescriptionPrefix: preparing graph")
 
     val shouldCheckpoint = checkpointInterval > 0
 
@@ -497,6 +530,7 @@ private[graphframes] object TwoPhase extends Logging {
 
     var lastRoundPersistedDFs = Seq[DataFrame](ee, minNbrs1)
     while (!converged) {
+      sc.setJobDescription(s"$jobDescriptionPrefix: iteration $iteration")
       var currRoundPersistedDFs = Seq[DataFrame]()
 
       // large-star step
@@ -603,6 +637,7 @@ private[graphframes] object TwoPhase extends Logging {
     logInfo(s"$logPrefix Connected components converged in ${iteration - 1} iterations.")
     logInfo(s"$logPrefix Join and return component assignments with original vertex IDs.")
 
+    sc.setJobDescription(s"$jobDescriptionPrefix: materializing final result")
     val output = buildOutput(graph, vv, ee, useLabelsAsComponents)
       .persist(intermediateStorageLevel)
 

diff --git a/core/src/main/scala/org/graphframes/mixins.scala b/core/src/main/scala/org/graphframes/mixins.scala
@@ -1,7 +1,58 @@
 package org.graphframes
 
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.storage.StorageLevel
 
+private[graphframes] object JobDescription {
+  // Mirrors SparkContext.SPARK_JOB_DESCRIPTION, which is private[spark].
+  private[graphframes] val JOB_DESCRIPTION_KEY = "spark.job.description"
+
+  /**
+   * Runs `body` and afterwards restores the Spark job description (a thread-local property) that
+   * the caller had set, so descriptions set inside `body` do not leak into jobs the caller
+   * triggers later on the same thread.
+   */
+  def withRestoredJobDescription[T](spark: SparkSession)(body: => T): T = {
+    val sc = spark.sparkContext
+    val previousDescription = sc.getLocalProperty(JOB_DESCRIPTION_KEY)
+    try {
+      body
+    } finally {
+      sc.setLocalProperty(JOB_DESCRIPTION_KEY, previousDescription)
+    }
+  }
+}
+
+/**
+ * Provides support for customizing the Spark job descriptions set by iterative algorithms.
+ *
+ * Job descriptions are shown in the "Description" column of the Jobs and Stages pages of the
+ * Spark UI, making the progress of long iterative runs visible without reading driver logs.
+ */
+private[graphframes] trait WithJobDescriptionPrefix {
+
+  /** The prefix used when no custom prefix is set, typically the algorithm name. */
+  protected def defaultJobDescriptionPrefix: String
+
+  protected var jobDescriptionPrefixOpt: Option[String] = None
+
+  /**
+   * Sets a custom prefix for the Spark job descriptions set by this algorithm (default: the
+   * algorithm name). Setting distinct prefixes allows telling apart multiple concurrent runs
+   * within the same Spark application.
+   */
+  def setJobDescriptionPrefix(value: String): this.type = {
+    jobDescriptionPrefixOpt = Some(value)
+    this
+  }
+
+  /**
+   * Gets the prefix of the Spark job descriptions set by this algorithm.
+   */
+  def getJobDescriptionPrefix: String =
+    jobDescriptionPrefixOpt.getOrElse(defaultJobDescriptionPrefix)
+}
+
 private[graphframes] trait WithAlgorithmChoice {
   protected val ALGO_GRAPHX = "graphx"
   protected val ALGO_GRAPHFRAMES = "graphframes"