Spark中的checkpoint用法

阳光穿透心脏的1/2处 2022-03-22 13:46 308阅读 0赞

怎么建立checkpoint  
首先需要用sparkContext设置hdfs的checkpoint的目录(如果不设置使用checkpoint会抛出异常:throw new SparkException(“Checkpoint directory has not been set in the SparkContext”):

scala> sc.setCheckpointDir("hdfs://lijie:9000/checkpoint0727")  
  
执行了上面的代码,hdfs里面会创建一个目录:

/checkpoint0727/c1a51ee9-1daf-4169-991e-b290f88bac20  
  
然后执行checkpoint

scala> val rdd1 = sc.parallelize(1 to 10000)  
rdd1: org.apache.spark.rdd.RDD\[Int\] = ParallelCollectionRDD\[0\] at parallelize at <console>:27

scala> rdd1.checkpoint  
  
发现hdfs中还是没有数据,通过collect然后hdfs就有数据了,说明checkpoint也是个transformation的算子

scala> rdd1.sum  
res2: Double = 5.0005E7

\#其中hdfs  
\[root@lijie hadoop\]\# hadoop dfs -ls /checkpoint0727/c1a51ee9-1daf-4169-991e-b290f88bac20/rdd-0  
DEPRECATED: Use of this script to execute hdfs command is deprecated.  
Instead use the hdfs command for it.

Found 2 items  
\-rw-r--r--   3 root supergroup      53404 2017-07-24 14:26 /checkpoint0727/c1a51ee9-1daf-4169-991e-b290f88bac20/rdd-0/part-00000  
\-rw-r--r--   3 root supergroup      53404 2017-07-24 14:26 /checkpoint0727/c1a51ee9-1daf-4169-991e-b290f88bac20/rdd-0/part-00001  
  
但是执行的时候相当于走了两次流程,sum的时候前面计算了一遍,然后checkpoint又会计算一次,所以一般我们先进行cache然后做checkpoint就会只走一次流程,checkpoint的时候就会从刚cache到内存中取数据写入hdfs中,如下:

rdd.cache()  
rdd.checkpoint()  
rdd.collect  
  
在checkpoint的时候强烈建议先进行cache,并且当你checkpoint执行成功了,那么前面所有的RDD依赖都会被销毁,如下:

/\*\*  
   \* Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint  
   \* directory set with \`SparkContext\#setCheckpointDir\` and all references to its parent  
   \* RDDs will be removed. This function must be called before any job has been  
   \* executed on this RDD. It is strongly recommended that this RDD is persisted in  
   \* memory, otherwise saving it on a file will require recomputation.  
   \*/