如何记住 Spark Streaming 应用程序重启之间的状态？答案

【问题标题】：How to remember state between Spark Streaming application's restarts?如何记住 Spark Streaming 应用程序重启之间的状态？
【发布时间】：2017-05-26 12:51:45
【问题描述】：

我正在学习 Spark Streaming。我想保持状态更新，并且能够使用 mapWithState 更新状态。我还在上下文中启用了检查点。如果我必须停止/重新启动工作，我想记住状态。现在每次重新启动都会重新开始计数。我尝试了各种现金、检查站选项，并扫描了很多张贴，但没有得到清晰的图片。

环境：我在本地开发中运行 Spark，也作为 HDP 沙箱运行。（我在两种环境中都试过）。

是否可以记住您终止 Spark 作业并重新启动它的状态。（无需任何编程更改）。
如果可能，怎么做？任何指示或建议都会有所帮助。（我已经尝试过检查点，在单个 RDD 上缓存，在本地以及在 HDP sanbox 中使用 MapwithStateRDD）。
我没有尝试的唯一选项是将 MapWithStateRDD 保存到磁盘并将其作为 initialRDD 读回。无论如何，感觉这不是正确的选择。

我只发现了一个没有答案的类似问题。 Spark Checkpoint doesn't remember state (Java HDFS)

谢谢。

代码：

  def getStreamingContext(streamingApp : (SparkContext, Duration) => StreamingContext, sc : SparkContext, batchDuration: Duration) = {
    val creatingFunc = () => streamingApp(sc, batchDuration)
    val ssc = sc.getCheckpointDir match {
      case Some(checkpointDir) =>
        println("Get or Create Context")
        StreamingContext.getActiveOrCreate(checkpointDir, creatingFunc, sc.hadoopConfiguration, createOnError = true)
      case None =>
        print("New Context")
        StreamingContext.getActiveOrCreate(creatingFunc)
    }
    sc.getCheckpointDir.foreach( cp => ssc.checkpoint(cp))
    println(ssc.getState())
    ssc
  }

Spark 版本 2.1.0

【问题讨论】：

使用检查点，但你说你启用了它......你能显示你初始化 StreamingContext 并设置检查点的代码吗？ Spark 版本是什么？
感谢您的快速回复 Jacek。
顺便说一句：我还尝试了一个课程材料中的练习，该课程材料具有完整的代码和检查点，并且表现相同。我想我可能必须将状态保存到 HDF 并在重启时初始化。
我得到了它的工作......感谢以下问答。 [link] (stackoverflow.com/questions/35675130/…) updateStateByKey statefulActivity.checkpoint(Minutes(1)) 解决了我的问题后，我错过了以下行。

标签： apache-spark spark-streaming

【解决方案1】：

我得到了它的工作...感谢以下问答。 [链接] (Spark streaming not remembering previous state)

我在 updateStateByKey 之后缺少以下行

statefulActivity.checkpoint(Minutes(1))

添加持续时间并启用检查点目录会记住重启时的状态。

【讨论】：