• Checkpointing
    • 何时checkpoint
    • 怎样配置Checkpointing

    Checkpointing

    一个流应用程序必须全天候运行，所有必须能够解决应用程序逻辑无关的故障（如系统错误，JVM崩溃等）。为了使这成为可能，Spark Streaming需要checkpoint足够的信息到容错存储系统中，
    以使系统从故障中恢复。

    • Metadata checkpointing：保存流计算的定义信息到容错存储系统如HDFS中。这用来恢复应用程序中运行worker的节点的故障。元数据包括
      • Configuration ：创建Spark Streaming应用程序的配置信息
      • DStream operations ：定义Streaming应用程序的操作集合
      • Incomplete batches：操作存在队列中的未完成的批
    • Data checkpointing ：保存生成的RDD到可靠的存储系统中，这在有状态transformation（如结合跨多个批次的数据）中是必须的。在这样一个transformation中，生成的RDD依赖于之前
      批的RDD，随着时间的推移，这个依赖链的长度会持续增长。在恢复的过程中，为了避免这种无限增长。有状态的transformation的中间RDD将会定时地存储到可靠存储系统中，以截断这个依赖链。

    元数据checkpoint主要是为了从driver故障中恢复数据。如果transformation操作被用到了，数据checkpoint即使在简单的操作中都是必须的。

    何时checkpoint

    应用程序在下面两种情况下必须开启checkpoint

    • 使用有状态的transformation。如果在应用程序中用到了updateStateByKey或者reduceByKeyAndWindow，checkpoint目录必需提供用以定期checkpoint RDD。
    • 从运行应用程序的driver的故障中恢复过来。使用元数据checkpoint恢复处理信息。

    注意，没有前述的有状态的transformation的简单流应用程序在运行时可以不开启checkpoint。在这种情况下，从driver故障的恢复将是部分恢复（接收到了但是还没有处理的数据将会丢失）。
    这通常是可以接受的，许多运行的Spark Streaming应用程序都是这种方式。

    怎样配置Checkpointing

    在容错、可靠的文件系统（HDFS、s3等）中设置一个目录用于保存checkpoint信息。着可以通过streamingContext.checkpoint(checkpointDirectory)方法来做。这运行你用之前介绍的
    有状态transformation。另外，如果你想从driver故障中恢复，你应该以下面的方式重写你的Streaming应用程序。

    • 当应用程序是第一次启动，新建一个StreamingContext，启动所有Stream，然后调用start()方法
    • 当应用程序因为故障重新启动，它将会从checkpoint目录checkpoint数据重新创建StreamingContext
    1. // Function to create and setup a new StreamingContext
    2. def functionToCreateContext(): StreamingContext = {
    3.     val ssc = new StreamingContext(...)   // new context
    4.     val lines = ssc.socketTextStream(...) // create DStreams
    5.     ...
    6.     ssc.checkpoint(checkpointDirectory)   // set checkpoint directory
    7.     ssc
    8. }
    10. // Get StreamingContext from checkpoint data or create a new one
    11. val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)
    13. // Do additional setup on context that needs to be done,
    14. // irrespective of whether it is being started or restarted
    15. context. ...
    17. // Start the context
    18. context.start()
    19. context.awaitTermination()

    如果checkpointDirectory存在，上下文将会利用checkpoint数据重新创建。如果这个目录不存在，将会调用functionToCreateContext函数创建一个新的上下文，建立DStreams。
    请看RecoverableNetworkWordCount例子。

    除了使用getOrCreate，开发者必须保证在故障发生时，driver处理自动重启。只能通过部署运行应用程序的基础设施来达到该目的。在部署章节将有更进一步的讨论。

    注意，RDD的checkpointing有存储成本。这会导致批数据（包含的RDD被checkpoint）的处理时间增加。因此，需要小心的设置批处理的时间间隔。在最小的批容量(包含1秒的数据)情况下，checkpoint每批数据会显著的减少
    操作的吞吐量。相反，checkpointing太少会导致谱系以及任务大小增大，这会产生有害的影响。因为有状态的transformation需要RDD checkpoint。默认的间隔时间是批间隔时间的倍数，最少10秒。它可以通过dstream.checkpoint
    来设置。典型的情况下，设置checkpoint间隔是DStream的滑动间隔的5-10大小是一个好的尝试。