Spark Streaming之检查点

喜欢ヅ旅行 2023-02-28 15:14 4阅读 0赞

## 一.缓存/持久化 ##

与RDD相似，DStream也允许开发人员将流的数据持久存储在内存中。也就是说，在DStream上使用persist()方法将自动对该DStream的每个RDD持久存储在内存中。如果DStream中的数据将被多次计算（例如，对同一数据进行多次操作），这将很有用。对于基于窗口操作reduceByWindow和reduceByKeyAndWindow以及基于状态操作的updateStateByKey而言，这都是隐含的。因此，由基于窗口操作生成的DStream会自动保存在内存中，而无需开发人员调用persist()。

对于通过网络接收数据的输入流（例如Kafka，Socket等），默认的持久性级别设置为将数据复制到两个节点以实现容错。

请注意，与RDD不同，DStream的默认持久性级别将数据序列化在内存中。

## 二.检查点 ##

流应用程序必须24/7全天候运行，因此必须对与应用程序逻辑无关的故障（例如系统故障，JVM崩溃等）具有弹性。为此，Spark Streaming需要将足够的信息检查点指向容错存储系统，以便可以从故障中恢复。检查点有两种类型的数据。

*  元数据检查点 -将定义流计算的信息保存到HDFS等容错存储中。这用于从运行流应用程序所在驱动程序的节点故障中恢复。元数据包括：
    
     *  配置 -用于创建流应用程序的配置。
     *  DStream操作 -定义流应用程序的DStream操作集。
     *  不完整的批次 -作业排队但尚未完成的批次。
 *  数据检查点 -将生成的RDD保存到可靠的存储中。在某些有状态转换中，这需要跨多个批次合并数据，这是必需的。在此类转换中，生成的RDD依赖于先前批次的RDD，这导致依赖项链的长度随时间不断增加。为了避免恢复时间的这种无限制的增加（与依赖性链成比例），有状态转换的中间RDD定期 检查点到可靠的存储（例如HDFS），以切断依赖性链。

总而言之，从驱动程序故障中恢复时，主要需要元数据检查点，而如果使用有状态转换，则即使是基本功能，也需要数据或RDD检查点。

**何时启用检查点**  
必须为具有以下任何要求的应用程序启用检查点：

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

注意，没有前述状态转换的简单流应用程序可以在不启用检查点的情况下运行。在这种情况下，从驱动程序故障中恢复也将是部分的（某些丢失但未处理的数据可能会丢失）。这通常是可以接受的，并且许多都以这种方式运行Spark Streaming应用程序。预计将来会改善对非Hadoop环境的支持。

**如何配置检查点**  
可以通过在容错，可靠的文件系统（例如HDFS，S3等）中设置目录来启用检查点，将检查点信息保存到该目录中。这是通过使用streamingContext.checkpoint(checkpointDirectory)来实现的。这将允许您使用前面提到的有状态转换。此外，如果要使应用程序从驱动程序故障中恢复，则应重写流应用程序以具有以下行为。

*  程序首次启动时，它将创建一个新的StreamingContext，设置所有流，然后调用star()。
 *  失败后重新启动程序时，它将根据检查点目录中的检查点数据重新创建StreamingContext。

/** * 创建ssc，根据检查点情况创建新的ssc * @return */
        def functionToCreateContext() : StreamingContext = { 
          val ssc = new StreamingContext(sc, Seconds(5))
          val lines = ssc.socketTextStream("master",9999) // 与nc端口对应
          val words = lines.flatMap(_.split(" "))
          var pairs = words.map(word=>(word,1)).reduceByKey(_+_)
          // 累加
          pairs = pairs.updateStateByKey[Int](updateFunction _) // 必须设置检查点
          pairs.foreachRDD(row => row.foreach(println))
          ssc.checkpoint(checkpointDirectory)
          ssc
        }
        // 根据检测点创建新的ssc
        val ssc = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)
        ssc.start()
        ssc.awaitTermination()
        ssc.stop()

如果checkpointDirectory存在，则将根据检查点数据重新创建上下文。如果该目录不存在（即首次运行），则将functionToCreateContext调用该函数以创建新上下文并设置DStreams。

除了使用getOrCreate驱动程序外，还需要确保驱动程序进程在发生故障时自动重新启动。这只能通过用于运行应用程序的部署基础结构来完成。

请注意，RDD的检查点会导致保存到可靠存储的成本。这可能会导致RDD获得检查点的那些批次的处理时间增加。因此，需要仔细设置检查点的间隔。在小批量（例如1秒）时，每批检查点可能会大大降低操作吞吐量。相反，检查点太不频繁会导致沿袭和任务规模增加，这可能会产生不利影响。对于需要RDD检查点的有状态转换，默认间隔为批处理间隔的倍数，至少应为10秒。可以使用 dstream.checkpoint(checkpointInterval)来设置。通常，DStream的5-10个滑动间隔的检查点间隔是一个不错的尝试。

## 三.累加器，广播变量和检查点 ##

无法从Spark Streaming中的检查点恢复累加器和广播变量。如果启用检查点并同时使用累加器或广播变量 ，则必须为累加器和广播变量创建延迟实例化的单例实例，以便在驱动程序发生故障重新启动后可以重新实例化它们。在下面的示例中显示。

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
      }
    }
    
    wordCounts.foreachRDD {  (rdd: RDD[(String, Int)], time: Time) =>
      // Get or register the blacklist Broadcast
      val blacklist = WordBlacklist.getInstance(rdd.sparkContext)
      // Get or register the droppedWordsCounter Accumulator
      val droppedWordsCounter = DroppedWordsCounter.getInstance(rdd.sparkContext)
      // Use blacklist to drop words and use droppedWordsCounter to count them
      val counts = rdd.filter {  case (word, count) =>
        if (blacklist.value.contains(word)) { 
          droppedWordsCounter.add(count)
          false
        } else { 
          true
        }
      }.collect().mkString("[", ", ", "]")
      val output = "Counts at time " + time + " " + counts
    })