[Spark]-Streaming-Persist与CheckPoint

1.Persist

　　与RDD类似,DStream允许用户将流数据存至内存或磁盘(依据用户设置的持久化级别).

　　注意:

　　　　在DStream中,默认是存储到内存中.(RDD默认MEMORY_ONLY)

　　　　对于基于窗口的操作, 如 reduceByWindow 和 reduceByKeyAndWindow 以及基于状态的操作, 如 updateStateByKey,内置是自动保存到内存的.

　　　　对于通过网络接收数据（例如: Kafka, Flume, sockets 等）的输入流, 默认持久性级别被设置为将数据复制到两个节点进行容错 (MEMORY_ONLY_2)

2.CheckPoint

　　2.1 概述

　　　　streaming 应用程序必须 24/7 运行, 因此必须对应用逻辑无关的故障（例如, 系统故障, JVM 崩溃等）具有弹性.所以Spark Streaming 需要 checkpoint 足够的信息到容错存储系统, 以便可以从故障中恢复

　　　　checkpoint 有两种类型的数据

　　　　　　Metadata checkpointing - 将定义 streaming 计算的信息保存到容错存储（如 HDFS）中.这用于从运行 streaming 应用程序的 driver 的节点的故障中恢复.Metadata包括:

　　　　　　　　Configuration 用于创建流应用程序的配置

　　　　　　　　DStream operations - 定义 streaming 应用程序的 DStream 操作集

　　　　　　　　ncomplete batches - 尚未完成的job 排队

　　　　　　Data checkpointing 生成的 RDD 保存到可靠的存储

　　　　　　　　在某些比如状态转换等的转换中,某些RDD是依赖先前的RDD.这可能会导致依赖链很长,CheckPoint可以将之后的RDD直接存储下来以切断这种依赖

　　2.2 使用

　　　　首先,为了凭借 CheckPoint 使 Streaming 应用从故障中恢复. Streaming 应用自身应该有从 CheckPoint 中创建StreamContext的能力

　　　　使用 StreamingContext.getOrCreate 可以简化这种行为,如下:　　　

        val checkPointPath = "D:\data\StreamingCheckpoint\StreamingApp"
        val ssc = StreamingContext.getActiveOrCreate(checkPointPath, ()=>{
            val conf = new SparkConf().setAppName("stream-app").setMaster("local[2]")
            val context = new StreamingContext(conf, Seconds(5))
            context.checkpoint(checkPointPath)
            context
        })

　　2.3 注意

　　　　2.3.1 累加器(Accumulators) 与广播变量(Broadcast Variables)

　　　　　　CheckPoint无法恢复累加器(Accumulators) 与广播变量(Broadcast Variables).

　　　　　　如果应用使用了累加器(Accumulators) 与广播变量(Broadcast Variables),需要为这两者创建延迟实例化的单例,以便Driver失败重启后重新实例化这两者　　　　　　　　　

 object WordBlacklist {      
        　　@volatile private var instance: Broadcast[Seq[String]] = null        
        　　def getInstance(sc: SparkContext): Broadcast[Seq[String]] = {
            　　if (instance == null) {
            　　　　synchronized {
                　　　　if (instance == null) {
                　　　　　　val wordBlacklist = Seq("a", "b", "c")
                　　　　　　instance = sc.broadcast(wordBlacklist)
                　　　　}
            　　　　}
            　　}
            　　instance
        　　}
        }
        
        object DroppedWordsCounter {       
        　　@volatile private var instance: LongAccumulator = null
        　　def getInstance(sc: SparkContext): LongAccumulator = {
           　　 if (instance == null) {
            　　　　synchronized {
                　　　　if (instance == null) {
                　　　　　　instance = sc.longAccumulator("WordsInBlacklistCounter")
                　　　　}
            　　　　}
            　　}
            　　instance
        　　}
        }

3.Persist与 CheckPoint 的区别　

　　Persist与 CheckPoint 这两者都可以将数据暂存让后面的任务使用.但是这两者还是有很大区别的.

　　最根本的区别在于这两者的定位完全不同. Persist定位是执行提速,而CheckPoint定位是容错. 其具体体现

　　i).数据容错

　　　　Persist 是允许保存到内存中,甚至允许在内存溢出时直接丢弃

　　　　CheckPoint 只能保存到磁盘.并且是本身支持容错的文件系统(HDFS)

　　ii).数据失效处理

　　　Persist 一旦driver关闭.Persist缓存数据会被清理掉.

　　　CheckPoint CheckPoint数据会永久保存

ii).执行机制

　　 Persist 是一种action.在执行Persist会立即产生计算并将计算结果写入到内存或磁盘

　　　 CheckPoint 是一个独立的Job.它会在某个计算任务完后,开启一个专门的任务来执行将结果保存.所以在CheckPoint中,计算会执行两次(所以建议rdd.cache().CheckPoint())