你有没有经历过凌晨三点被电话叫醒，打开监控一看：Kafka Lag 以肉眼可见的速度往上飙，SparkStreaming 作业的 Driver 已经不知道什么时候挂掉了？如果你还没有给 Driver 做 HA（High Availability），所有所谓的“自动恢复”就都是空谈。等人工重启之后你会发现数据断了几小时，下游报表全部错位，这一晚上的数据链路基本等于瘫痪。
这篇文章把 SparkStreaming 的 Driver HA 掰开揉碎讲清楚：为什么 Driver 是流任务的命门、Checkpoint 恢复机制底层到底存了什么、ZooKeeper 与资源框架分别起到什么作用，以及最终如何基于 getOrCreate 和一组参数在生产环境把 Driver HA 真正落起来。目标读者是已经在用 SparkStreaming 做实时计算，但还没有给作业做高可用保护、或做了但恢复链路不清晰的工程师。
1. 深夜断流与单点命门：先搞懂Driver在Streaming里扛着什么
很多新手把 SparkStreaming 的 HA 简单理解为“给任务配个重启脚本”，这个认知是危险的。不搞清楚 Driver 在 Streaming 作业里具体管着什么，你搭出来的所谓 HA 很可能只是在重启一个不断丢失状态的空壳。
Driver 进程在 SparkStreaming 作业里承载的职责比普通 Spark 批作业更重。它不仅仅是 SparkContext 的宿主编排 Spark 任务，还要管理实时计算特有的运行时状态。以最常见的 Kafka Direct 模式为例，Driver 至少要扛住这么几件事：
- DStreamGraph 的构建与维护：这个数据流图记录了每个 DStream 之间的依赖关系、窗口操作逻辑、状态操作的算子链路，是整个流计算的处理蓝图。
- JobScheduler 的运行：每个 batch interval 会触发一个或多个 Spark Job，这些 Job 的生成、排队、提交和状态跟踪都运行在 Driver 端。
- InputDStream 的 offset 管理：Kafka Direct 模式下，每次读取哪些分区的哪些 offset，是由 Driver 端当前保存的 offset 决定的。Driver 挂了，这个消费位置信息也就暂时丢了，除非落到了 Checkpoint 或外部系统中。
- ReceivedBlockTracker 的状态：如果你用的是 Receiver 方式，Receiver 接收到的数据块分配给了哪个 batch、每个 batch 对应了哪些 Block，这些信息同样记录在 Driver 端。
一句话总结：Streaming 作业的“大脑”就在 Driver 里。大脑死掉，即使 Executor 全活着、Kafka 的数据还在堆积，作业也无法继续调度。更麻烦的是，没有 HA 保护的作业在 Driver 挂掉后，整个作业进程往往直接消失，不会自动拉起来。数据从挂在那一刻开始断流，直到有人发现并重新提交作业。
理解了这一点，你就能明白为什么 Driver HA 是整个 SparkStreaming 高可用方案里优先级最高的一环。它要解决的其实是两个问题：
- 容器/进程层面：Driver 进程挂了，有没有外部机制把它重新拉起来？
- 状态恢复层面：进程重新拉起之后，能不能从挂掉的瞬间继续跑，而不是从头消费数据或直接丢掉 Checkpoint 前的处理结果？
这两个问题缺一不可。只解决进程拉起，不解决状态恢复，拉起来之后数据可能从 Kafka earliest 重新消费，造成海量重复计算；只解决状态恢复，不解决进程拉起，那恢复逻辑写得再漂亮也没人执行。
2. Driver恢复的地基：Checkpoint元数据备份到底存了什么
掉旧坑之前先把地基打牢。SparkStreaming 的 Driver HA 核心依赖是 Checkpoint，它的本质是定期把 Driver 端的元数据序列化后写到可靠的共享文件系统（生产环境一般就是 HDFS）。恢复的时候，新的 Driver 进程从这个目录反序列化出刚才提到的 DStreamGraph、未完成 batch、block 分配信息等，从而重建出完整的运行现场。
很多人以为 Checkpoint 就是“存了一份 RDD 数据”，其实不对。SparkStreaming 的 Checkpoint 分为两种，用途完全不同：
| Checkpoint 类型 | 存储内容 | 主要用途 |
|---|---|---|
| Metadata Checkpoint | SparkConf 配置、DStreamGraph 逻辑、未完成 Batch 的元数据、Receiver 接收块的分配信息 | 恢复 Driver 运行框架 |
| Data Checkpoint | 带状态算子（如 updateStateByKey、reduceByKeyAndWindow）产生的中间 RDD 数据 | 恢复跨 batch 的计算状态 |
Metadata Checkpoint 里最核心的是 DStreamGraph 的序列化结果。它保存的不是“代码逻辑”，而是运行过程中已经构建好的 DStream 对象图——包括每个 InputDStream 当前消费到了哪个 offset、每次操作对应的函数对象、每个状态算子的配置。所以恢复时不依赖重新执行一遍代码来构建图，而是直接加载这个对象图。
Data Checkpoint 则专门服务于跨 batch 的状态计算。比如你维护了一个按天累计的 key 计数，状态必须存在某个位置才能在下一个 batch 继续累加。实时状态默认在内存中，Driver 一挂内存就没了，因此 SparkStreaming 会按一定周期把状态 RDD 也 Checkpoint 到 HDFS。这个周期称为 checkpoint interval，默认等于 batch interval，但生产上一般建议设置为 batch interval 的 5 到 10 倍，避免频繁写 HDFS 拖垮处理性能。
恢复流程拆开看是这样的：
- 新的 Driver 进程启动，通过
StreamingContext.getOrCreate检测到 Checkpoint 目录下存在有效的元数据文件。 - 框架从 HDFS 反序列化 Metadata Checkpoint，恢复 SparkConf 和 DStreamGraph。
- 基于恢复出的图，重新创建 StreamingContext 的内部组件，比如 JobScheduler、ReceiverTracker。
- 找到上次尚未处理完成的 batch，从那个时间点重新生成 Job 并提交执行。
- Data Checkpoint 中的状态 RDD 会作为状态算子的初始状态加载回去，保证跨 batch 累计不中断。
所以你看到的“自动续跑”不是魔法，而是这套元数据恢复机制在背后兜底。但这里也有一个极其容易踩的坑：Checkpoint 里存的是 DStreamGraph 对象，不是代码。如果你修改了业务代码里的 DStream 转换逻辑，从旧 Checkpoint 恢复时加载出来的仍然是旧的图结构，新的代码逻辑根本不会生效。反之，如果删掉 Checkpoint 目录再启动，新代码才会通过 creatingFunc 重新构建。
3. 谁来把Driver拉起来：ZooKeeper与资源框架的角色分工
Checkpoint 负责的是“恢复状态”，但“把 Driver 拉起来”这件事本身，需要外部机制介入。这就引出了两种常见部署模式下的不同实现方式：Standalone 集群模式和 YARN 模式。
3.1 Standalone 模式：ZooKeeper 管集群，supervise 管 Driver
在 Spark 自带的 Standalone 集群里，如果只启动一个 Master，那么 Master 本身也是单点。所以要让 Driver 能被重新拉起，先要让 Master 先具备 HA 能力，常见做法是把 Master 的元数据放到 ZooKeeper 里，让多个 Master 节点通过 ZK 选主，active Master 挂掉后 standby Master 自动接管。
这时候 Spark 的 Driver HA 实际上是两层配合：
spark.deploy.recoveryMode=ZOOKEEPER让 Master 把自己的状态（包括已注册的 Driver 信息）持久化到 ZooKeeper。- 提交作业时加上
--supervise，Master 才会在 Driver 进程异常退出后考虑重新调度一个 DriverWrapper 到其他存活 Worker 上。
拉起后的 Driver 进程仍然是一个全新的 JVM 进程。它启动后执行的还是你提交的那个 main 类中的代码，所以代码里必须有StreamingContext.getOrCreate逻辑，否则新进程只会重新创建一个全新的 StreamingContext，不会去读 Checkpoint，更不会自动续跑。
Standalone 模式下恢复链路可以这样描述：Worker 上的 Driver 进程挂掉（机器宕机或 OOM）→ Master 通过 ZK 中保存的信息感知 Driver 丢失 → 检查该 Driver 是否设置了 supervise → 在可用 Worker 上重新调度容器启动新 Driver → 新 Driver 从 HDFS Checkpoint 恢复 StreamingContext → 作业续跑。
3.2 YARN 模式：利用 ApplicationMaster 重试机制
大多数公司的生产集群都在 YARN 上，这种模式下 Driver 实际上运行在 ApplicationMaster（AM）内部。Spark 提交到 YARN 的任务本身就具备 AM 重启的能力：只要 AM 异常退出，ResourceManager 会根据spark.yarn.max.app.attempts配置判断是否重新调度一个 AM 容器。
这个设计天然适合做 Driver HA，因为重启后的新 AM 容器里会重新启动 Driver 进程。你只需要保证：
spark.yarn.max.app.attempts大于 1，否则 AM 挂掉直接就是任务失败，不会重试。- Checkpoint 目录在共享存储上，新 AM 里的 Driver 能正常读取。
- 代码里正确使用
getOrCreate。
YARN 模式比 Standalone 省心的地方在于你不需要单独维护一套 Spark Master 的 HA，YARN ResourceManager 本身就是高可用的，它会妥善处理调度和重试。需要额外注意的一点是：YARN 上的 AM 重试如果是由于“内存超限”“OOM Kill”等被 NodeManager 杀死的情况，新的 AM 会被调度到其他节点，但旧的容器可能残留一段时间。因此 Spark 内部会有防止多个 AM 同时处理同一个作业的机制，这也是官方建议不要在 YARN 模式下手动频繁 kill AM 来测试的原因之一。
4. 落地实操：getOrCreate与关键参数，一步步搭出Driver HA
理论部分讲完，下面直接进入可复制的落地步骤。我以 Kafka Direct + Scala 为例，实际项目里改动最大的就是这个入口类和一堆 spark-submit 参数。
4.1 代码侧：必须用 getOrCreate 包装 StreamingContext
直接用new StreamingContext的作业没有办法享受 Driver 恢复能力。正确姿势是让启动逻辑被StreamingContext.getOrCreate接管：
import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.kafka010._ def createStreamingContext(): StreamingContext = { val conf = new SparkConf().setAppName("DriverHAExample") val ssc = new StreamingContext(conf, Seconds(5)) // 这里根据你的数据源构造 DStreamGraph val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "kafka1:9092,kafka2:9092", "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer", "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer", "group.id" -> "driver-ha-demo", "auto.offset.reset" -> "earliest", "enable.auto.commit" -> "false" ) val topics = Array("demo_topic") val stream = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) ) stream.map(_.value()) .map(record => (record.split(",")(0), 1L)) .reduceByKeyAndWindow(_ + _, _ - _, Seconds(60), Seconds(10)) .foreachRDD { rdd => // 你的业务处理逻辑，建议做幂等写入 rdd.foreachPartition { iter => // 写外部存储 } } ssc.checkpoint("hdfs://nameservice/user/spark/checkpoint/driver-ha-demo") ssc } def main(args: Array[String]): Unit = { val checkpointPath = "hdfs://nameservice/user/spark/checkpoint/driver-ha-demo" val ssc = StreamingContext.getOrCreate(checkpointPath, createStreamingContext _) ssc.sparkContext.setLogLevel("WARN") ssc.start() ssc.awaitTermination() }这段代码里有几个细节值得抠一下：
ssc.checkpoint的路径必须放在共享文件系统上，本地盘路径在 Driver 换节点后读不到。StreamingContext.getOrCreate的行为是：Checkpoint 目录存在且恢复成功 → 直接加载元数据；不存在 → 调用传入的createStreamingContext函数创建新作业。所以检查点目录一旦存在，创建函数里的 DStream 构建逻辑不会重新执行。- 恢复成功后不要再次调用
ssc.checkpoint来覆盖路径，路径以 Checkpoint 文件里的记录为准。 enable.auto.commit=false是为了不依赖 Kafka 消费者组管理 offset，否则恢复时的 offset 管理会和 Checkpoint 冲突。
4.2 Standalone 模式提交：--supervise 是关键
如果跑在 Standalone 集群，spark-submit 要带着--supervise参数，同时给 Spark 配好 ZK：
spark-submit \ --class com.example.DriverHAExample \ --master spark://master1:7077,master2:7077 \ --deploy-mode cluster \ --supervise \ --conf spark.deploy.recoveryMode=ZOOKEEPER \ --conf spark.deploy.zookeeper.url=zk1:2181,zk2:2181,zk3:2181 \ --conf spark.deploy.zookeeper.dir=/spark-ha \ --conf spark.streaming.kafka.maxRatePerPartition=1000 \ driver-ha-demo.jar这里--supervise是最容易被漏掉的一项。不加它，Spark Master 也会感知到 Driver 退出，但不会重新调度。加上之后，Master 才会在你当前作业的 submitted driver 列表里维护一个需要重启的状态。
4.3 YARN 模式提交：调大 AM 重试次数
YARN 模式不需要--supervise，核心是下面几个参数：
spark-submit \ --class com.example.DriverHAExample \ --master yarn \ --deploy-mode cluster \ --conf spark.yarn.max.app.attempts=5 \ --conf spark.yarn.am.attemptFailuresValidityInterval=1h \ --conf spark.streaming.kafka.maxRatePerPartition=1000 \ driver-ha-demo.jarspark.yarn.max.app.attempts默认值是 1，不显式调大，AM 挂掉直接任务失败。我见过不少团队在这里吃过亏，辛辛苦苦搭好了 Checkpoint，结果 AM 重试参数没改，Driver 一挂作业就进入 FAILED 状态，恢复逻辑完全没机会执行。
另外要注意 YARN 的硬性限制：spark.yarn.max.app.attempts不能超过 ResourceManager 侧的yarn.resourcemanager.am.max-attempts，后者通常默认是 2。如果 RM 侧没调大，你这里写 5 也会被强制压回 2。
4.4 常用参数速查与说明
| 参数 | 推荐值 | 说明 |
|---|---|---|
| spark.deploy.recoveryMode | ZOOKEEPER | 仅 Standalone 模式，开启 Spark Master 元数据持久化 |
| spark.deploy.zookeeper.url | zk1:2181,zk2:2181 | Master 元数据存储的 ZK 地址 |
| spark.deploy.zookeeper.dir | /spark-ha | ZK 中存储 Spark 元数据的 znode 路径 |
| spark.yarn.max.app.attempts | 3 到 5 | YARN 模式 AM 最大重试次数，决定 Driver 能自动拉起几次 |
| spark.yarn.am.attemptFailuresValidityInterval | 1h | 统计 AM 失败时间窗口，避免历史失败累积导致不再重试 |
| spark.streaming.receiver.writeAheadLog.enable | true | Receiver 方式下开启 WAL，减少数据丢失概率 |
| spark.streaming.receiver.writeAheadLog.rollingFile.maxSize | 512MB | 控制 WAL 文件滚动大小，减少 HDFS 小文件 |
| spark.streaming.stopSparkContextByDefault | false | 恢复后 Driver 退出时是否连带关闭 SparkContext |
这里额外解释一下spark.streaming.stopSparkContextByDefault。它默认是 true，含义是 StreamingContext 停止时自动联动停止 SparkContext。但在 Driver HA 场景下，如果新 Driver 从 Checkpoint 恢复失败、或某个批处理抛出致命异常导致 ssc.stop()，这个默认行为会让 SparkContext 一起退出，最终把 AppMaster 也带崩溃。生产上建议显式设为 false，至少保住 SparkContext，方便排查和进一步处理。
4.5 验证 Driver HA 是否真的生效
搭完之后一定要做一次故障演练，不要等到线上真的挂了才发现配置不对。实践中我一般按这个顺序验证：
- 先把作业正常提交，确认 Checkpoint 目录在 HDFS 上生成，能看到
metadata子目录里有文件写入。 - 登录 YARN Web UI 找到当前 ApplicationMaster 的运行节点和容器 ID。
- 在对应节点上执行
kill -9杀掉 AM 进程，模拟最极端的崩溃场景。 - 观察 YARN 是否在若干秒后自动调度新的 AM。
- 查看新 AM 的日志，正常情况下会看到
Recovered from checkpoint之类的日志，说明恢复流程已经触发。 - 再确认 Kafka 消费位点不是从 earliest 重新开始，而是从挂掉前的位置继续。
如果发现恢复后作业确实从 Checkpoint 续跑，但处理 Lag 明显高于常态，这是正常现象，说明恢复过程中积压的 batch 正在被补算。不要慌，等它追到实时进度即可。
5. 恢复不是保险箱：WAL、幂等消费与那些易踩的坑
Driver HA 能自动重启进程并恢复元数据，但这绝不等于数据处理“既不丢也不重”。实际操作中，很多从外表看已经开了 HA 的作业，在数据语义上仍然有隐患。
5.1 WAL 到底在防什么
如果你是老派 Receiver 方式，数据先由 Receiver 接收并存放在 Executor 内存中，Driver 端只记录块元数据。这个过程有个天然弱点：Receiver 收到了数据，但还没等 Driver 记录“这个 batch 包含哪些 Block”，Driver 挂了，恢复后这些 Block 对应关系丢失，数据也就丢了。WAL（Write Ahead Log）要解决的就是这个时间窗问题：Receiver 收到数据后先写入 HDFS 上的 WAL，再更新内存中的块信息。恢复时可以从 WAL 重放数据。
Kafka Direct 模式其实不太依赖 WAL，因为 offset 就保存在 Checkpoint 或外部存储里，数据源本身就是可重放的。但如果你混合使用了 Receiver 和 Kafka Direct，或任务里有其他第三方数据源，开启 WAL 仍然是一个重要的安全垫。
5.2 重复消费与幂等是必须面对的现实
这里必须泼一盆冷水：即使配置全部到位，Driver HA 提供的也只是 At-Least-Once 语义，不是 Exactly-Once。原因很简单：恢复时那个“未完成 batch”到底执行到哪一步，Spark 并没有精确记录。有可能 Job 已经跑完了大部分任务，只是结果还没来得及写入外部存储，Driver 就挂了。恢复后这个 batch 会被重新调度执行，下游就会收到重复数据。
所以生产落地的铁律是：下游写入必须做幂等或幂等控制。比如写 MySQL，用唯一键插入或 update 而不是无条件 insert；写 ES，用带 id 的 upsert；写 Kafka，给消息加全局唯一 ID，消费端自己做去重。没有幂等保护，HA 恢复的往往是“半截数据”，比没有 HA 更让人头疼。
5.3 Checkpoint 与代码变更的冲突
前面提过一句，这里值得展开。Checkpoint 恢复的是序列化后的 DStreamGraph 对象，不是当前运行代码。这意味着：
- 如果你改了 DStream 的转换逻辑（比如从 map 改成 flatMap，或者改了窗口大小），从旧 Checkpoint 恢复时，运行的是旧逻辑，代码变更完全不会生效。
- 如果你改了类名、包名、算子里引用的类结构，反序列化时还可能直接报 ClassNotFoundException 或序列化异常，恢复失败。
- 新逻辑要上线的标准做法是：要么清空旧的 Checkpoint 目录重新开始消费，要么另开一个全新的 Checkpoint 路径。无论如何，都要接受重新消费已有数据或位点重置带来的业务影响。
我踩过的坑是有一次只换了一个工具类的内部实现，没改类名，结果反序列化时旧对象带出来的字段结构和新类对不上，恢复直接报错。后来吸取的教训是：凡是涉及代码逻辑变更，先主动验证 Checkpoint 能恢复，不能恢复就果断清理检查点。
5.4 Checkpoint 间隔与 HDFS 小文件
Checkpoint 写得越频繁，Driver 恢复时丢失的状态越少，但代价是 HDFS 上会产生大量小文件，同时对 NameNode 造成压力。Metadata Checkpoint 每个 batch 都会写一次默认情况下；Data Checkpoint 则按 checkpoint interval 触发。
生产调优建议：Metadata Checkpoint 频率保持默认即可，Data Checkpoint 间隔按状态数据量和恢复容忍度调整，一般设为 batch interval 的 5 到 10 倍。如果你的 batch 是 5 秒，那 checkpoint interval 可以设 30 秒左右；如果 batch 是 1 分钟，那 5 到 10 分钟都可以。写 HDFS 前也可以考虑用压缩：
ssc.checkpoint("hdfs://nameservice/...") ssc.conf.set("spark.rdd.compress", "true")spark.rdd.compress能在一定程度上减少 Checkpoint 和 shuffle 中间数据的体积，但对 CPU 有额外开销，要结合集群资源情况权衡。
5.5 Kafka Direct 模式 offset 恢复的两个细节
第一个细节：恢复后auto.offset.reset=earliest不会因为 Checkpoint 里已经保存了 offset 而重新从最早消费。这个配置只在没有 Checkpoint 且 Kafka 里没有已提交 offset 时才会生效。所以不要在恢复失败后天真地以为调小 earliest 就能把数据捡回来。
第二个细节：如果你同时启用了 Kafka 自身的 offset 提交（enable.auto.commit=true），那么 Checkpoint 和 Kafka 记录的 offset 可能不一致。SparkStreaming 官方建议在 streaming 场景下禁用 Kafka 自动提交，以 Checkpoint 为准。否则恢复时两边 offset 打架，会出现跳过数据或重复消费的奇怪现象。
5.6 同机恢复与资源不足
虽然 YARN 和 Standalone 都会尽量把新 Driver 调度到其他节点，但如果集群资源紧张，新的 AM 容器有可能被分配到同一台故障率较高的机器上。更麻烦的是，Driver 恢复时需要重新创建 SparkContext，如果之前的 Executor 动态分配策略没配好，恢复后可能长期只有少量 Executor 在处理积压 batch。为此可以配合spark.dynamicAllocation.enabled=true和spark.dynamicAllocation.maxExecutors来给恢复过程留一些弹性。
6. 我这几年的实操心得与验证建议
最后聊一点个人经验，不一定能写进官方文档，但都是真实趟过后觉得值得记下来的点。
第一，能上 YARN 尽量上 YARN，别在 Standalone 上硬磕。Standalone 的 Driver HA 虽然原理上很清晰，但它要求你额外维护 Spark Master 的 ZK HA，而且--supervise之后新 Driver 经常被调度到任意一个 Worker 上，客户端日志采集、监控 Agent 部署都得跟着适配。YARN 模式下 AM 重启是资源管理器原生能力，配合 Checkpoint 就能把恢复链路走通，运维负担小很多。
第二，把故障演练当成上线流程的一部分。我每搭好一个新的 Streaming 任务，都会在测试环境做一次“杀 Driver 进程”演练，确认日志里出现从 Checkpoint 恢复的记录，确认 Kafka Lag 能在几分钟内回落。很多参数问题只有真刀真枪杀进程才暴露得出来。演练脚本很简单，核心就是根据 YARN applicationId 找到 AM 所在节点，然后 kill 对应进程。跑通一次之后，心里才真正有底。
第三，监控不能只看 Kafka Lag，还要看 Driver 重启事件。很多 HA 配置能自动恢复，但恢复过程通常有几分钟空窗，数据会积压。如果监控维度太粗，可能业务方已经发现报表延迟了，你还在因为 Lag 还没突破阈值而没收到告警。建议对 YARN AM 重启次数、SparkStreaming 的 job 提交时间间隔做单独告警，这两个指标能第一时间反映 Driver 是否发生了重建。
第四，永远为幂等做好准备。即使你觉得当前下游系统不会产生重复数据，也要在设计阶段把幂等逻辑加上。因为 Driver HA 恢复的“未完成 batch”在极端情况下可能被重复执行多次，如果下游是无脑 append，数据质量会直接崩掉。这个成本在开发阶段看似多余，但在真实故障发生时是最能救命的一道防线。
SparkStreaming 的 Driver HA 不是一个开关就能解决的问题，它需要代码侧、参数侧、部署侧三条线同时配合。把 getOrCreate 用对、把 AM 重试次数调大、把 Checkpoint 放对地方、把下游幂等做好，这一套走下来，你才算是真正给流任务上了一道保险。
