Checkpointing is a process of truncating RDD lineage graph and saving it to a reliable distributed (HDFS) or local file system.
There are two types of checkpointing:
-
reliable - in Spark (core), RDD checkpointing that saves the actual intermediate RDD data to a reliable distributed file system, e.g. HDFS.
-
local - in Spark Streaming or GraphX - RDD checkpointing that truncates RDD lineage graph.
It’s up to a Spark application developer to decide when and how to checkpoint using RDD.checkpoint() method.
Before checkpointing is used, a Spark developer has to set the checkpoint directory using SparkContext.setCheckpointDir(directory: String) method.
You call SparkContext.setCheckpointDir(directory: String) to set the checkpoint directory - the directory where RDDs are checkpointed. The directory must be a HDFS path if running on a cluster. The reason is that the driver may attempt to reconstruct the checkpointed RDD from its own local file system, which is incorrect because the checkpoint files are actually on the executor machines.
You mark an RDD for checkpointing by calling RDD.checkpoint(). The RDD will be saved to a file inside the checkpoint directory and all references to its parent RDDs will be removed. This function has to be called before any job has been executed on this RDD.
|
Note
|
It is strongly recommended that a checkpointed RDD is persisted in memory, otherwise saving it on a file will require recomputation. |
When an action is called on a checkpointed RDD, the following INFO message is printed out in the logs:
15/10/10 21:08:57 INFO ReliableRDDCheckpointData: Done checkpointing RDD 5 to file:/Users/jacek/dev/oss/spark/checkpoints/91514c29-d44b-4d95-ba02-480027b7c174/rdd-5, new parent is RDD 6When RDD.checkpoint() operation is called, all the information related to RDD checkpointing are in ReliableRDDCheckpointData.
spark.cleaner.referenceTracking.cleanCheckpoints (default: false) - whether clean checkpoint files if the reference is out of scope.
Beside the RDD.checkpoint() method, there is similar one - RDD.localCheckpoint() that marks the RDD for local checkpointing using Spark’s existing caching layer.
This RDD.localCheckpoint() method is for users who wish to truncate RDD lineage graph while skipping the expensive step of replicating the materialized data in a reliable distributed file system. This is useful for RDDs with long lineages that need to be truncated periodically, e.g. GraphX.
Local checkpointing trades fault-tolerance for performance.
The checkpoint directory set through SparkContext.setCheckpointDir is not used.