zoukankan      html  css  js  c++  java
  • [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())

  • 相关阅读:
    STL——(4)deque
    Windows Server服务器日常管理技巧
    各种服务器伪静态配置方法
    DedeCMS V5.7开启memcache缓存的方法配置说明
    windows服务器下iis的性能优化 服务器
    网站出现service unavailable的解决方法
    【ThinkPHP框架学习 】(2) --- 后台管理系统如何用iframe点击左边右边局部刷新
    【ThinkPHP框架学习 】(1) --- thinkphp 3.2.3 验证码验证使用教程分享
    走着走着,已经到了玩不起的年龄 —— 致自己
    PHP常用数组(Array)函数整理
  • 原文地址:https://www.cnblogs.com/NightPxy/p/9311521.html
Copyright © 2011-2022 走看看