0


Spark性能优化三 checkpoint

(一)checkpoint介绍

checkpoint,是Spark提供的一个比较高级的功能。有时候,我们的Spark任务,比较复杂,从初始化RDD开始,到最后整个任务完成,有比较多的步骤,比如超过10个transformation算子。而且,整个任务运行的时间也特别长,比如通常要运行1~2个小时。
在这种情况下,就比较适合使用checkpoint功能了。
因为对于特别复杂的Spark任务,有很高的风险会出现某个要反复使用的RDD因为节点的故障导致丢失,
虽然之前持久化过,但是还是导致数据丢失了。那么也就是说,出现失败的时候,没有容错机制,所以当后面的transformation算子,又要使用到该RDD时,就会发现数据丢失了,此时如果没有进行容错处理的话,那么就需要再重新计算一次数据了。
所以针对这种Spark Job,如果我们担心某些关键的,在后面会反复使用的RDD,因为节点故障导致数据丢失,那么可以针对该RDD启动checkpoint机制,实现容错和高可用。

(二)使用checkpoint

  1. 首先要调用SparkContext的setCheckpointDir()方法,设置一个容错的文件系统的目录,比如HDFS;然后,对RDD调用checkpoint()方法。
  2. 最后,在RDD所在的job运行结束之后,会启动一个单独的job,将checkpoint设置过的RDD的数据写入之前设置的文件系统中。
  3. 这是checkpoint使用的基本步骤,很简单,那我们下面先从理论层面分析一下当我们设置好checkpoint之后,Spark底层都做了哪些事情。在这里插入图片描述

(三)checkpoint与持久化的区别

  1. lineage是否发生改变 linage(血缘关系)说的就是RDD之间的依赖关系 持久化,只是将数据保存在内存中或者本地磁盘文件中,RDD的lineage(血缘关系)是不变的; Checkpoint执行之后,RDD就没有依赖的RDD了,也就是它的lineage改变了
  2. 丢失数据的可能性 持久化的数据丢失的可能性较大,如果采用 persist 把数据存在内存中的话,虽然速度最快但是也是最不可靠的,就算放在磁盘上也不是完全可靠的,因为磁盘也会损坏。 Checkpoint的数据通常是保存在高可用文件系统中(HDFS),丢失的可能性很低

(四)checkpoint的使用

object CheckpointOpScala {def main(args: Array[String]):Unit={val conf =new SparkConf()
    conf.setAppName("CheckpointOpScala").setMaster("local")val sc =new SparkContext(conf)if(args.length==0){
      System.exit(100)}//1:摄者checkpoint目录
    sc.setCheckpointDir("hdfs:bigdata01:9000/chk001")val dataRdd = sc.textFile("hdfs://bigdata01:9000/hello.txt")//2:对rddcheckpoint 操作
    dataRdd.checkpoint()val ouputPath = args(0)
    dataRdd.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile(ouputPath)
    sc.stop()}privatedef getSparkContext ={val conf =new SparkConf()
    conf.setAppName("CreateRddByArrayscala").setMaster("local")new SparkContext(conf)}}

本文转载自: https://blog.csdn.net/qq_40905284/article/details/129310869
版权归原作者 csm_81 所有, 如有侵权,请联系我们删除。

“Spark性能优化三 checkpoint”的评论:

还没有评论