You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
docs: randomized_contraction always needs a checkpoint directory (#909)
useLocalCheckpoints removes the need for a checkpoint directory for two_phase (TwoPhase.scala:271), but randomized_contraction always requires one (RandomizedContraction.scala:63-75): it writes each iteration's representatives under that directory and reads them back by path.
For `two_phase` and `randomized_contraction`. By default, GraphFrames uses persistent checkpoints, which are reliable but require a `checkpointDir` to be configured in persistent storage (e.g. S3 or HDFS). Setting `use_local_checkpoints=True` uses the local disks of Spark executors instead. Local checkpoints are faster but less reliable: if an executor is lost, the checkpoint is lost and the job will fail.
334
+
For `two_phase` and `randomized_contraction`. By default, GraphFrames uses persistent checkpoints, which are reliable but require a `checkpointDir` to be configured in persistent storage (e.g. S3 or HDFS). Setting `use_local_checkpoints=True` uses the local disks of Spark executors instead. Local checkpoints are faster but less reliable: if an executor is lost, the checkpoint is lost and the job will fail.`randomized_contraction` still requires a `checkpointDir`, because it writes its intermediate tables there.
|`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, which removes the need to configure a checkpoint directory. Local checkpoints are faster but less reliable: they do not survive the loss of an executor. Used by the "two_phase" and "randomized_contraction" algorithms. | Optional (default: `false`) | 0.9.3 |
11
+
|`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 "two_phase" this removes the need to configure a checkpoint directory. "randomized_contraction" 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 "two_phase" and "randomized_contraction" algorithms. | Optional (default: `false`) | 0.9.3 |
12
12
|`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](/04-user-guide/05-traversals.md) for the details. Used by the "two_phase" and "randomized_contraction" algorithms. | Optional (default: `false`) | 0.9.0 |
13
13
|`spark.graphframes.connectedComponents.algorithm`| Sets the connected components algorithm to use. Supported algorithms: <br>- "two_phase": Uses alternating large star and small star iterations proposed in [Connected Components in MapReduce and Beyond](http://dx.doi.org/10.1145/2670979.2670997). <br>- "randomized_contraction": Uses the randomized algorithm proposed in [In-database connected component analysis](https://arxiv.org/pdf/1802.09478). <br>- "graphx": Converts the graph to a GraphX graph and then uses the connected components implementation in GraphX. <br>- "graphframes": Deprecated alias for "two_phase". | Optional (default: `two_phase`) | 0.9.0 |
14
14
|`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 |
15
-
| `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 and local checkpoints are not enabled, this throws a `java.io.IOException`. Set a nonpositive value to disable checkpointing. Used by the "two_phase" and "randomized_contraction" 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 |
15
+
| `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, "randomized_contraction" throws a `java.io.IOException`, and so does "two_phase" unless local checkpoints are enabled. Set a nonpositive value to disable checkpointing. Used by the "two_phase" and "randomized_contraction" 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 |
16
16
|`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 |
17
17
18
18
## Setting Configurations
@@ -107,7 +107,7 @@ result.show()
107
107
108
108
## Notes on Configuration Usage
109
109
110
-
-**Checkpoint Directory**: For configurations related to checkpointing, make sure to set a checkpoint directory using `spark.sparkContext.setCheckpointDir("path/to/checkpoint/dir")` before running algorithms that use checkpointing. Where `sparkContext` is not reachable, for example from a Spark Connect client, set the `spark.checkpoint.dir` configuration instead; GraphFrames falls back to it. Alternatively, enable `spark.graphframes.useLocalCheckpoints`, which needs no checkpoint directory at all.
110
+
-**Checkpoint Directory**: For configurations related to checkpointing, make sure to set a checkpoint directory using `spark.sparkContext.setCheckpointDir("path/to/checkpoint/dir")` before running algorithms that use checkpointing. Where `sparkContext` is not reachable, for example from a Spark Connect client, set the `spark.checkpoint.dir` configuration instead; GraphFrames falls back to it. For "two_phase" you can instead enable `spark.graphframes.useLocalCheckpoints`, which needs no checkpoint directory. "randomized_contraction" always needs one.
111
111
-**Storage Levels**: When setting the `spark.graphframes.connectedComponents.intermediatestoragelevel` configuration, use one of the following values: `MEMORY_ONLY`, `MEMORY_AND_DISK`, `MEMORY_ONLY_SER`, `MEMORY_AND_DISK_SER`, `DISK_ONLY`, `MEMORY_ONLY_2`, `MEMORY_AND_DISK_2`, etc.
112
112
-**Algorithm Selection**: The choice of algorithm for connected components can significantly impact performance. The DataFrame-based algorithms, "two_phase" (the default) and "randomized_contraction", are generally more scalable for large graphs, while the RDD-based "graphx" algorithm may be faster for smaller graphs. See [Traversals](/04-user-guide/05-traversals.md) for a comparison of the algorithms.
113
-
-**Local Checkpoints**: Local checkpoints are faster and can put strain on the local disk if insufficiently large. They are also less reliable than checkpoints in persistent storage, because they do not survive the loss of an executor. Because local checkpoints do not require setting `checkpointDir`, they are a convenient option for shorter runs.
113
+
-**Local Checkpoints**: Local checkpoints are faster and can put strain on the local disk if insufficiently large. They are also less reliable than checkpoints in persistent storage, because they do not survive the loss of an executor. Because local checkpoints do not require setting `checkpointDir` for "two_phase", they are a convenient option for shorter runs.
0 commit comments