Spark 檢查點（checkpoint）
什么是Checkpointing
Checkpointing可以將RDD從其依賴關系中抽出來，保存到可靠的存儲系統(tǒng)（例如HDFS，S3等)， 即它可以將數據和元數據保存到檢查指向目錄中。 因此，在程序發(fā)生崩潰的時候，Spark可以恢復此數據，并從停止的任何地方開始。
Checkpointing分為兩類：
- 高可用checkpointing，容錯性優(yōu)先。這種類型的檢查點可確保數據永久存儲，如存儲在HDFS或其他分布式文件系統(tǒng)上。 這也意味著數據通常會在網絡中復制，這會降低檢查點的運行速度。
- 本地checkpointing，性能優(yōu)先。 RDD持久保存到執(zhí)行程序中的本地文件系統(tǒng)。 因此，數據寫得更快，但本地文件系統(tǒng)也不是完全可靠的，一旦數據丟失，工作將無法恢復。一般用于需要定期截取且擁有較長的lineage關系的RDD，例如，GraphX。
開發(fā)人員可以是來RDD.checkpoint()方法來設置檢查點。在使用檢查點之前，必須使用SparkContext.setCheckpointDir(directory: String)方法設置檢查點目錄。
所以其實我們的checkpoint主要用于Spark Streaming任務
為什么使用Checkpointing
RDD的檢查點機制就好比Hadoop將中間計算值存儲到磁盤，即使計算中出現了故障，我們也可以輕松地從中恢復。通過對 RDD 啟動檢查點機制可以實現容錯和高可用。
- 在Spark Streaming程序中，如果某些數據已經在隊列中等待處理，由于某些原因我們的應用程序崩潰，當我們再次啟動時，則無需再次讀取這些數據，并且數據不會丟失。
- 如果我們的應用程序正在使用任何有狀態(tài)操作，那么檢查點是必需的，否則一旦應用程序崩潰，所有狀態(tài)都將丟失。
哪些RDD需要使用Checkpointing
- 計算需要很長時間的
- 計算鏈太長的
- 依賴于太多的父RDD
Cache、Persist和Checkpoint的區(qū)別
cache()與persist()的區(qū)別
會被重復使用的但是不能太大的RDD需要cache。cache()調用了persist()，區(qū)別在于cache只有一個默認的緩存級別MEMORY_ONLY，而persist可以根據情況設置其它的緩存級別，StorageLevel類中有12種緩存級別。
cache機制是每計算出一個要cache的partition就直接將其cache到內存了。但checkpoint沒有使用這種第一次計算得到就存儲的方法，而是等到job結束后另外啟動專門的job去完成checkpoint ，也就是說需要checkpoint的RDD會被計算兩次。因此在使用rdd.checkpoint()的時候建議加上rdd.cache()，這樣第二次運行的 job 就不用再去計算該rdd了，直接讀取cache寫磁盤。
persist()與checkpoint()的區(qū)別
rdd.persist(StorageLevel.DISK_ONLY) 與 checkpoint 也有區(qū)別。前者雖然可以將RDD的partition持久化到磁盤，但該partition由blockManager管理。一旦driver program執(zhí)行結束，也就是executor所在進程CoarseGrainedExecutorBackend結束了，blockManager也會相應退出，被 cache 到磁盤上的 RDD 也會被清空，整個blockManager使用的local文件夾被刪除。
而checkpoint將RDD持久化到HDFS或本地文件夾，如果不被手動remove掉，是一直存在的，也就是說可以被下一個driver program使用，而cached RDD不能被其他dirver program使用。
建立CheckPointing示例
用sparkContext設置hdfs的checkpoint的目錄。
scala> sc.setCheckpointDir("hdfs:/tmp/checkpoint")
利用上面代碼建立好檢查點目錄后，hdfs的會出現類似下面的目錄。
[dev@test06 ~]$ hdfs dfs -ls /tmp/checkpoint
Found 1 items
drwxr-xr-x - hadoop supergroup 0 2019-04-30 10:50 /tmp/checkpoint/b4282eb3-cde8-489b-afda-4f1d08b9c236
執(zhí)行檢查點
scala> val rdd1=sc.parallelize(1 to 1000)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[11] at parallelize at <console>:24
scala> rdd1.checkpoint
這個時候在hdfs的/tmp/checkpoint/b4282eb3-cde8-489b-afda-4f1d08b9c236這個目錄下是找不到任何數據的。但是通過collect后，這個目錄就有數據了,說明checkpoint也是個transformation的算子。文章來源：http://www.zghlxwxcb.cn/news/detail-859355.html
scala> rdd1.sum
res6: Double = 500500.0
[dev@test06 ~]$ hdfs dfs -ls /tmp/checkpoint/b4282eb3-cde8-489b-afda-4f1d08b9c236
Found 1 items
drwxr-xr-x - ccpgdev supergroup 0 2019-04-30 10:57 /tmp/checkpoint/b4282eb3-cde8-489b-afda-4f1d08b9c236/rdd-11
像上面說的，由于對RDD設置檢查點的時候，需要對RDD進行兩次計算，所以建議在設置checkpointing之前，先對rdd調用cache()進行緩存起來，避免重復計算同一個rdd。文章來源地址http://www.zghlxwxcb.cn/news/detail-859355.html
scala> rdd1.cache()
res8: rdd1.type = ParallelCollectionRDD[11] at parallelize at <console>:24
scala> rdd1.checkpoint()
scala> rdd1.collect()
res10: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125, 126, 127, 128, 129, 130, 131, 132, 133, 134, 135, 136, 137, 138, 139, 140, 141, 142, 143, 144, 145, 146, 147, 148, 149, 150, 151, 152, 153, 154, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172, 173, 174, 175, 176,...
scala>
到了這里，關于Spark 檢查點（checkpoint）的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網！
