★ FEATURED ARTICLE 1. 为什么流计算里“记住状态”比想象中难流计算和离线批处理最大的区别在于数据是源源不断进来的没有“跑完这一批就结束”的概念。离线任务挂了重跑一遍就行反正输入数据是固定的但流任务挂了如果重启后从零开始消费要么重复计算要么丢失中间状态业务上根本没法接受。这就是checkpoint 机制存在的根本原因——它要解决的是“故障恢复后从哪里接着算、状态怎么还原”这两个问题。Spark Streaming 的 checkpoint 不是单一功能而是一套组合拳包含两个层面的东西一是元数据检查点Metadata Checkpointing把作业的配置、DStream 的血缘关系、未完成的批次信息存下来用于驱动重启二是数据检查点Data Checkpointing把算子产生的中间状态比如 updateStateByKey 维护的累计值持久化用于状态恢复。很多人第一次接触时会把这两者混为一谈结果配了 checkpoint 目录却发现状态还是丢了问题往往就出在这里。这篇文章适合正在用 Spark Streaming 做实时统计、实时累计、会话分析这类有状态计算的开发者也适合那些“作业跑得好好的一重启数据就乱套”的排查者。我会从设计思路讲到实操配置再到踩过的坑尽量把每个参数背后的“为什么”说清楚让你看完能直接对着自己的作业改。2. checkpoint 机制的整体设计与选型考量2.1 两个检查点到底各管什么先把这个概念掰开。Spark Streaming 的 checkpoint 目录里实际上会存几类东西它们的作用完全不同存储内容归属作用丢失后果元数据配置、DStream 图Metadata Checkpoint驱动重启时重建作业无法用 Checkpoint 恢复只能重写代码未完成批次Metadata Checkpoint重启后接着处理批次丢失或重复RDD 状态数据Data Checkpoint恢复有状态算子的中间值状态归零累计值从头算已处理偏移部分场景依赖外部存储定位消费位置重复消费或漏消费关键点在于元数据检查点是为了让 Driver 能“复活”数据检查点是为了让状态能“续命”。如果你只用了无状态算子map、filter 这类理论上只需要元数据检查点就能恢复但一旦用了updateStateByKey、mapWithState、窗口聚合这些有状态操作数据检查点就必须开否则重启后状态是空的。2.2 为什么 Spark 选择“血缘 检查点”双保险Spark 的核心设计哲学是 RDD 血缘Lineage靠血缘可以重算丢失的分区这是它容错的基础。但流计算里血缘有个致命问题血缘会无限增长。一个跑了三天的作业DStream 的依赖链可能长到几百层一旦某个 RDD 丢失重算代价高得离谱甚至根本算不回来因为上游数据可能已经不在 Kafka 里了。所以 Spark Streaming 的策略是用检查点把血缘“截断”。定期把状态 RDD 落盘之后如果出错直接从最近的检查点恢复而不是从头重算。这就像打游戏存档——你不会每次都从第一关重打而是从最近的存档点继续。检查点间隔就是这个“存档频率”存太勤影响性能存太稀恢复时重算量大需要权衡。2.3 选型时绕不开的几个决策实际配置时有几个决策点必须提前想清楚检查点目录放哪本地文件系统只适合测试生产必须用可靠的分布式存储如 HDFS 或对象存储因为 Driver 可能换机器重启本地盘上的检查点就找不到了。是否用 getOrCreate这是官方推荐的 Driver 重启方式它会在检查点目录存在时自动恢复不存在时新建。很多人手动 new StreamingContext结果重启后配置对不上直接报错。检查点间隔怎么定官方建议是批次间隔的 5 到 10 倍。批次 1 秒检查点就 5 到 10 秒批次 1 分钟检查点就 5 到 10 分钟。太短会让每次落盘都成为负担太长则恢复时重算的数据量太大。提示检查点间隔不是越小越安全。每次检查点都会触发一次同步写盘如果状态很大写盘时间可能超过批次处理时间反而拖垮整个作业。3. 核心细节解析与实操要点3.1 检查点目录的结构长什么样当你设置好ssc.checkpoint(hdfs://.../checkpoint)后Spark 会在该目录下生成一套结构。理解这个结构排查问题时能少走很多弯路checkpoint/ ├── metadata # 元数据文件记录批次信息和 DStream 图 ├── batchTime.rdd # 状态 RDD 数据有状态算子才有 ├── batchTime.rdd # 每个检查点批次一个 └── ...metadata文件是核心它记录了最后一个成功检查点的批次时间、未完成的批次队列、以及 DStream 的序列化图。重启时getOrCreate会读这个文件重建 StreamingContext。如果这个文件损坏或丢失作业就无法从检查点恢复只能重新开始。状态 RDD 文件则是按批次时间命名的比如1700000000000.rdd。恢复时Spark 会找到最近的一个状态 RDD加载它作为初始状态然后从对应的批次继续消费。3.2 有状态算子的检查点行为差异不同的有状态算子检查点的行为并不一样这点很容易被忽略updateStateByKey每个批次都会把全量状态写进检查点。状态越大写盘越慢。如果 key 数量上百万检查点开销会非常可观。mapWithState内部用了更高效的 MapWithStateRDD检查点只存增量变化的部分性能比 updateStateByKey 好很多。新项目建议优先用它。窗口操作窗口内的数据会被保留在检查点里窗口越长、批次越密检查点体积越大。我实测过一个场景用 updateStateByKey 维护 50 万个 key 的累计值批次 10 秒检查点 1 分钟每次检查点写盘要 3 到 5 秒。换成 mapWithState 后同样的状态量写盘降到 1 秒以内。这个差距在状态规模大时非常明显。3.3 配置检查点的正确姿势下面是一段可以直接参考的配置骨架重点看注释里的说明val conf new SparkConf() .setAppName(StatefulStreamingJob) // 序列化用 Kryo状态对象序列化更快、体积更小 .set(spark.serializer, org.apache.spark.serializer.KryoSerializer) // 注册自定义状态类避免 Kryo 每次写类名 .registerKryoClasses(Array(classOf[MyState])) val ssc StreamingContext.getOrCreate( checkpointDir, // 检查点目录生产用 HDFS () createContext(conf) // 首次创建时的逻辑 ) // 关键先设置检查点再定义有状态算子 ssc.checkpoint(checkpointDir) // 批次间隔 10 秒 val stream KafkaUtils.createDirectStream(...) // 有状态计算 val stateDstream stream.updateStateByKey[Long](updateFunc) ssc.start() ssc.awaitTermination()这里有几个容易踩的点ssc.checkpoint()必须在定义有状态算子之前调用否则会报“checkpoint directory not set”的错误。getOrCreate的第二个参数是一个函数只在检查点不存在时执行。如果你把创建逻辑写在函数外面重启时会重复创建导致端口冲突或资源浪费。用 Kryo 序列化时自定义的状态类必须注册否则每次序列化都会带上完整类名检查点文件会膨胀。注意getOrCreate恢复时会用检查点里的配置覆盖你代码里的配置。如果你改了批次间隔或并行度重启后可能不生效需要先删掉检查点目录重新跑。这是很多人“改了配置没反应”的根因。4. 实操过程与核心环节实现4.1 从零搭一个有状态作业并验证恢复光看配置不够得实际跑一遍才能理解恢复流程。下面是我常用的验证步骤用一个词频累计的例子走通全流程。第一步准备一个本地或 HDFS 的检查点目录比如/tmp/ssc-checkpoint。测试阶段用本地目录就行生产换成 HDFS 路径。第二步写一个带updateStateByKey的作业逻辑很简单从 socket 或 Kafka 读单词累计每个单词出现的次数。def updateFunc(newValues: Seq[Long], runningCount: Option[Long]): Option[Long] { val newCount runningCount.getOrElse(0L) newValues.sum Some(newCount) } val lines ssc.socketTextStream(localhost, 9999) val words lines.flatMap(_.split( )) val pairs words.map(word (word, 1L)) val runningCounts pairs.updateStateByKey[Long](updateFunc) runningCounts.print()第三步启动作业往 socket 端口发几批数据观察控制台输出的累计值。比如发三次 a b应该看到 a 和 b 的计数从 1 涨到 3。第四步手动杀掉进程模拟故障。然后重新启动同一个作业。这时候关键点来了如果检查点配置正确重启后 a 和 b 的计数应该从 3 继续累加而不是从 0 开始。如果从 0 开始说明数据检查点没生效状态丢了。第五步再发一批 a如果输出是 4说明恢复成功如果输出是 1说明状态没恢复检查点目录或配置有问题。这个验证过程看起来简单但能帮你把“元数据恢复”和“状态恢复”两件事都验证到。我建议每个新作业上线前都跑一遍这个流程比事后排查省事得多。4.2 检查点间隔的计算与调优检查点间隔不是拍脑袋定的有个经验公式可以参考检查点间隔 max(批次间隔 × 5, 状态写盘耗时 × 3)举个例子批次间隔 10 秒状态写盘耗时 2 秒那么检查点间隔取 max(50, 6) 50 秒。也就是说至少 50 秒存一次档保证写盘不会成为瓶颈。但如果你对恢复时的数据重复很敏感可以适当缩短。比如金融类累计宁可多写几次盘也要把恢复时的重算窗口压小。这时候可以取批次间隔的 3 倍代价是写盘频率上升。还有一个隐藏参数spark.streaming.checkpoint.interval在代码里是通过ssc.checkpoint(dir)配合批次间隔隐式决定的但你可以通过ssc.checkpoint(dir)后跟ssc.graph.checkpointInterval来显式设置。不过更常见的做法是直接调ssc.checkpoint(dir)并接受默认的批次倍数关系简单可靠。4.3 Driver 重启的完整流程拆解理解 Driver 重启时发生了什么对排查问题特别有帮助。整个流程大致是这样作业启动调用StreamingContext.getOrCreate。检查点目录存在且metadata文件有效进入恢复分支。读取metadata反序列化 DStream 图重建 StreamingContext。找到最近的状态 RDD 文件加载为初始状态。从metadata里记录的未完成批次开始重新提交这些批次。继续正常消费。这里面第 5 步是重复消费的高发区。因为未完成批次会被重新处理如果你的输出端比如数据库写入不是幂等的就会出现重复数据。解决办法有两个一是输出端做幂等比如用主键 upsert二是用事务性输出Spark 2.2 之后支持幂等写入的 sink。提示metadata文件里的未完成批次队列长度是有限的默认会保留最近几个批次。如果作业挂了很久才重启超过保留窗口的批次可能已经不在队列里恢复时会跳过它们造成数据丢失。所以故障后要尽快重启别拖。5. 常见问题与排查技巧实录5.1 重启后状态归零的几种原因这是最高频的问题我整理了一张速查表现象可能原因排查方法状态从 0 开始检查点目录配置错误确认ssc.checkpoint()路径与getOrCreate一致状态从 0 开始有状态算子定义在 checkpoint 之前调整代码顺序先 checkpoint 再定义算子状态从 0 开始检查点目录被清空检查 HDFS 目录是否有.rdd文件报错找不到类状态类未注册 Kryo添加registerKryoClasses恢复后配置不对检查点覆盖了新配置删掉检查点目录重新跑我遇到过一次很隐蔽的情况检查点目录写的是相对路径本地测试没问题上了集群后 Driver 换机器相对路径解析到了不同位置状态自然找不到。后来统一改成绝对路径的 HDFS 地址才解决。所以检查点路径一定要用绝对路径生产环境用分布式存储。5.2 检查点写盘拖慢作业怎么办如果发现批次处理时间突然变长且和检查点间隔吻合大概率是写盘成了瓶颈。可以从几个方向优化换 mapWithState如果还在用 updateStateByKey优先迁移。mapWithState 的增量检查点机制能大幅降低写盘量。压缩状态状态对象里只存必要字段别把整个原始记录塞进去。我见过有人把整条 JSON 存进状态检查点文件几个 G纯属自找麻烦。调大检查点间隔在可接受的恢复重算范围内把间隔拉长。提升存储带宽HDFS 写入慢的话考虑换更高吞吐的存储或者增加并行度。实测下来把 updateStateByKey 换成 mapWithState再配合状态瘦身检查点写盘耗时能降 60% 以上。这个优化性价比很高值得花时间做。5.3 检查点目录清理的坑检查点目录会随着时间不断增长每个批次的状态 RDD 都存着不清理的话磁盘很快爆掉。Spark 本身有自动清理机制默认保留最近几个检查点但有时候会失效尤其是作业频繁重启时。我的做法是加一个独立的清理任务定期扫描检查点目录删掉超过一定时间的历史文件。但要注意不能删metadata文件和最近的状态 RDD否则恢复会失败。清理逻辑要保守宁可多留几个也别误删。另外如果作业要改逻辑重新上线最稳妥的方式是换一个新的检查点目录而不是在旧目录上覆盖。旧目录留着做回滚备份新目录干净启动避免新旧状态混在一起出问题。5.4 一个容易被忽略的序列化问题状态对象如果包含不可序列化的字段比如数据库连接、文件句柄检查点时会直接抛异常。这类问题在测试时不一定暴露因为测试数据量小可能还没触发检查点就结束了。解决办法是让状态类只包含纯数据字段把连接这类资源放到算子外部管理或者用transient标记。我踩过一次坑状态类里存了一个 SimpleDateFormat序列化时报NotSerializableException排查了半天才发现是这个。后来改成在 updateFunc 内部临时创建问题消失。注意Kryo 序列化虽然快但对某些类型支持不完善。如果遇到奇怪的序列化错误可以先切回 Java 序列化验证是不是 Kryo 的问题确认后再针对性注册类。6. 关于检查点我个人的几条实战体会跑了几年流作业关于 checkpoint 我最大的体会是它不是一个配了就完事的开关而是一套需要持续关注的基础设施。很多团队上线时配了检查点之后就不管了直到某天作业挂了重启才发现状态恢复不了数据全乱。我的建议是把检查点健康度纳入日常监控。具体可以盯几个指标检查点写盘耗时、检查点目录大小增长趋势、最近一次成功检查点的时间。如果写盘耗时持续上升说明状态在膨胀该考虑优化了如果目录大小增长异常快可能是状态设计有问题。另外定期做恢复演练。别等到真出故障才第一次验证恢复流程那时候手忙脚乱容易出错。我一般每个月挑一个低峰期手动重启一次作业确认状态能正常恢复。这个习惯帮我提前发现过好几次配置漂移的问题。最后分享一个小技巧在metadata文件里其实记录了作业的批次信息你可以写个小脚本定期解析它监控未完成批次的数量。如果这个数字持续大于 0说明作业处理速度跟不上输入速度迟早会出问题。这个信号比等作业报警要早得多能给你留出充足的调整时间。
阅读完成 · 觉得有帮助？
