Configurations

GraphFrames provides several configuration options that can be used to tune the behavior of algorithms and operations. This page documents all available configurations, their descriptions, default values, and usage examples.

Configuration Table

The following table lists all available GraphFrames configurations:

Configuration Description Default Value Since Version
spark.graphframes.useLocalCheckpoints Tells the connected components algorithm to use local checkpoints. If set to "true", the algorithm checkpoints to the local disks of the executors instead of to persistent storage. For "twophase" this removes the need to configure a checkpoint directory. "randomizedcontraction" always needs one, because it stores its intermediate tables there. Local checkpoints are faster but less reliable: they do not survive the loss of an executor. Used by the "twophase" and "randomizedcontraction" algorithms. Optional (default: false) 0.9.3
spark.graphframes.useLabelsAsComponents Tells the connected components algorithm to use the minimum original vertex label as the component ID. When "false", component IDs have the data type LONG. The effect depends on the algorithm and on the vertex ID type, see Traversals for the details. Used by the "twophase" and "randomizedcontraction" algorithms. Optional (default: false) 0.9.0
spark.graphframes.connectedComponents.algorithm Sets the connected components algorithm to use. Supported algorithms:
- "twophase": Uses alternating large star and small star iterations proposed in Connected Components in MapReduce and Beyond.
- "randomizedcontraction": Uses the randomized algorithm proposed in In-database connected component analysis.
- "graphx": Converts the graph to a GraphX graph and then uses the connected components implementation in GraphX.
- "graphframes": Deprecated alias for "two_phase".		 Optional (default: two_phase) 0.9.0
spark.graphframes.connectedComponents.broadcastthreshold Sets broadcast threshold in propagating component assignments. If a node degree is greater than this threshold at some iteration, its component assignment will be collected and then broadcasted back to propagate the assignment to its neighbors. Otherwise, the assignment propagation is done by a normal Spark join. Set it to -1 to disable manual broadcasting and let Adaptive Query Execution handle the skew instead; that mode is considerably faster on most graphs. This parameter is only used by the "two_phase" algorithm. Optional (default: 1000000) 0.9.0
spark.graphframes.connectedComponents.checkpointinterval Sets checkpoint interval in terms of number of iterations. Checkpointing regularly helps recover from failures, clean shuffle files, shorten the lineage of the computation graph, and reduce the complexity of plan optimization. As of Spark 2.0, the complexity of plan optimization would grow exponentially without checkpointing. Hence, disabling or setting longer-than-default checkpoint intervals are not recommended. Checkpoint data is saved under the directory set by org.apache.spark.SparkContext.setCheckpointDir, or, when that is not set, under the spark.checkpoint.dir configuration; the algorithm writes into a sub-directory prefixed with its own name ("connected-components" or "randomized-contraction"). If neither is set, "randomizedcontraction" throws a java.io.IOException, and so does "twophase" unless local checkpoints are enabled. Set a nonpositive value to disable checkpointing. Used by the "twophase" and "randomizedcontraction" algorithms; note that "two_phase" in AQE mode (broadcastthreshold set to -1) relies on Spark's own DataFrame checkpointing instead of its own sub-directory. Optional (default: 2) 0.9.0
spark.graphframes.connectedComponents.intermediatestoragelevel Sets storage level for intermediate datasets that require multiple passes. Used by all the connected components algorithms. Optional (default: MEMORY_AND_DISK) 0.9.0

Setting Configurations

GraphFrames configurations can be set in several ways:

Spark Configuration

You can set configurations when creating a SparkSession:

Scala API

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("GraphFrames Example")
  .config("spark.graphframes.connectedComponents.algorithm", "two_phase")
  .config("spark.graphframes.connectedComponents.checkpointinterval", 3)
  .getOrCreate()

Python API

from pyspark.sql import SparkSession

spark = SparkSession.builder \
  .appName("GraphFrames Example") \
  .config("spark.graphframes.connectedComponents.algorithm", "two_phase") \
  .config("spark.graphframes.connectedComponents.checkpointinterval", 3) \
  .getOrCreate()

Runtime Configuration

You can also set configurations at runtime:

Scala API

spark.conf.set("spark.graphframes.connectedComponents.algorithm", "two_phase")
spark.conf.set("spark.graphframes.connectedComponents.checkpointinterval", 3)

Python API

spark.conf.set("spark.graphframes.connectedComponents.algorithm", "two_phase")
spark.conf.set("spark.graphframes.connectedComponents.checkpointinterval", 3)

Example: Connected Components with Custom Configurations

This example shows how to run the Connected Components algorithm with custom configurations:

Scala API

import org.graphframes.GraphFrame
import org.graphframes.examples

// Get example graph
val g = examples.Graphs.friends

// Set configurations
spark.conf.set("spark.graphframes.connectedComponents.algorithm", "two_phase")
spark.conf.set("spark.graphframes.connectedComponents.checkpointinterval", 3)
spark.conf.set("spark.graphframes.useLocalCheckpoints", true)

// Run connected components with custom configurations
val result = g.connectedComponents.run()
result.show()

Python API

from graphframes.examples import Graphs

# Get example graph
g = Graphs(spark).friends()

# Set configurations
spark.conf.set("spark.graphframes.connectedComponents.algorithm", "two_phase")
spark.conf.set("spark.graphframes.connectedComponents.checkpointinterval", 3)
spark.conf.set("spark.graphframes.useLocalCheckpoints", "true")

# Run connected components with custom configurations
result = g.connectedComponents()
result.show()

Notes on Configuration Usage