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重新消费数据即可。
