spark streaming checkpointing 踩坑

spark streaming的应用可能需要7*24小时不间断的运行,因此需要一定的容错能力。在系统出现问题后，spark streaming 应用能够从上次出错的地方重新开始。为此spark streaming提供了checkpointing机制来应对该问题。

checkpoint在实现时，需要保持两类数据：

1 Metadata checkpointing

当运行streaming应用的node节点出现故障时，就需要使用元数据的checkpointing数据进行恢复。这里的元数据包括：

配置 创建该流时的各项配置信息

DStream的操作 该流用到的各项DStream相关操作

未完成的批次 可能有批次已经进入到队列中，但是尚未完成，需要记录相关信息。

2 Data checkpointing

保存RDD数据。 在stateful DStream中，会合并跨批次的RDD数据。合并出的RDD依赖于之前批次的RDD。在这种情况下，用checkpoint来切断依赖链路。关于有状态DStream 详见《spark streaming stateful DStream 持久保存RDD/有状态的内存》

1 checkpoint 的使用：

streamingContext的创建要使用getOrCreate方法，主要需要将streaming的相关处理逻辑都放到该方法中。示例如下

object StreamApp extends Logging {
  def createContext(taskId: String, streamConf: StreamConf): StreamingContext = {
    val exit_code = 0
    val sparkConf = new SparkConf()
    sparkConf.set("spark.scheduler.mode", "FAIR")
    sparkConf.setAppName(s"halcyon ${taskId}")

    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(streamConf.getBatch_interval))
    ssc.checkpoint(streamConf.getStateCheckpointPath)
    try {
      StreamProcess.process(ssc, streamConf)
    } catch {
      case interruptException: InterruptedException => {
        logInfo("Streaming stopped")
      }
      case e: Exception => {
        e.printStackTrace()
        logWarning("Streaming exit due to Exception :" + e.getMessage)
      }
      case err: Error =>  {
        err.printStackTrace()
        logWarning("Streaming exit due to Error:" + err.getMessage)
      }
    } finally {
      logInfo("system exit code: " + exit_code)
      if (exit_code !=0 ) {
        throw new Exception("user exception for retry")
      }
    }
    ssc
  }

  def main(args: Array[String]) {

    if (args.length < 1) {
      System.exit(1)
    }
    val taskId = args(0)
    System.setProperty("STREAM_LOG_PATH", CommonConstant.logPath)
    val streamConf = new StreamConf
    val ssc =
      if (streamConf.getStreamCheckpointData)
        StreamingContext.getOrCreate(streamConf.getStateCheckpointPath, () => createContext(taskId, streamConf))
      else
        createContext(taskId, streamConf)
    ssc.start()
    if (StreamingContextState.ACTIVE == ssc.getState()) {
      logInfo("Start halcyon " + taskId + " success !")
    }
    ssc.awaitTermination()
  }
}

由于checkpoint需要将数据写入到持久存储中，会影响批次处理的时间。选择一个合适的checkpoint时间较为重要。默认是batch时间的倍数，最小10s。官方的推荐是5-10个批次进行一次checkpoint

2 checkpoint的缺陷：

1 代码更新后checkpoint数据不可用

checkpoint实现中将Scala/Java/Python objects序列化存储起来，恢复时会尝试反序列化这些objects。如果用修改过的class可能会导致错误。此时需要更换checkpoint目录或者删除checkpoint目录中的数据，程序才能起来。

2 spark1.6 中对dataframe的支持有限

在spark1.6中，对dataframe进行checkpoint可能会无法恢复。从spark2.1开始对dataframe checkpoint有好的支持见issuse：https://issues.apache.org/jira/browse/SPARK-11879

有一些trick的方法使用，可能会成功，详见：
https://stackoverflow.com/questions/33424445/how-to-checkpoint-dataframes/37014202#37014202对rdd进行checkpoint，再创建dataframe可以成功。
笔者在Stateful DStream中使用上述方法仍然失败。

总结：

由于checkpoint天生的缺陷即代码变更后不能进行恢复，在生产环境中由于程序潜在的不稳定、程序的升级，checkpoint的缺陷都会造成一定风险。在这里不推荐使用checkpoint。可以参考《Spark Streaming 容错机制》。

需要注意的是当没有使用checkpoint时可能造成数据丢失的情况，即该流从数据源拉取了数据但是未来记得处理就发生了故障，这部分数据会丢失。所以，在不实用checkpoint时，比如数据来源是kafka，我们可以保存消费kafka的offset，当出现上述情况时，流重新拉起后，从上次的offset重新消费数据即可。