Spark Streaming中的检查点 Checkpoint简介；如何使用检查点存储/恢复处理结果

水深无声 2021-11-01 12:10 562阅读 0赞

# 检查点 Checkpoint功能 #

1.  保存 每批中 state信息，累加加销售营业额
2.  保存 没冲从Kafka topic 中读取数据的offset
3.  保存DStream的来源和DStream处理函数和输出函数

# 什么时候需要使用Checkpoint #

有状态转换的用法 - 如果在应用程序中使用了**updateStateByKey**或**reduceByKeyAndWindow**（**with inverse function**），则必须提供检查点目录以允许定期RDD检查点。  
从运行应用程序的驱动程序的故障中恢复 - 元数据检查点用于使用进度信息进行恢复。

# 使用检查点存储恢复处理结果 #

## 设置检查点 ##

val CHECK_POINT_PATH =“filepath:///”
    def functionToCreateContext(): StreamingContext = {
      val ssc = new StreamingContext(...)   // new context
      val lines = ssc.socketTextStream(...) // create DStreams
      ...
      ssc.checkpoint(CHECK_POINT_PATH)   // set checkpoint directory
      ssc
    }

## 使用检查点 ##

def main(args: Array[String]): Unit = {
     val ssc: StreamingContext = StreamingContext.getOrCreate(
          CHECK_POINT_PATH,
          createFunc
        )
     def createFunc():StreamingContext={
          val conf = new SparkConf()
            .setMaster("local[3]") //为什么启动3个，有一个Thread运行Receiver
            .setAppName("F_CheckPointOrderTotalStreaming")
          val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
          //日志级别
          ssc.sparkContext.setLogLevel("WARN")
    
          processData(ssc)
          ssc.checkpoint(CHECK_POINT_PATH)
          ssc
        }
       }