Spark Structured Streaming 状态膨胀与 Checkpoint 故障排查

Spark Structured Streaming 长时间运行后状态数据持续膨胀导致 OOM，或者 checkpoint 目录损坏无法恢复，根本原因取决于查询是 stateless 还是 stateful。两种场景的状态膨胀机制完全不同，对应的排查方向和恢复策略也截然不同。区分清楚这一点，才能避免在错误的方向上浪费时间。

stateless 与 stateful 的状态膨胀差异

先明确一个概念：checkpoint 目录中保存的不只是"状态数据"，还包括 offset 信息、WAL（预写日志）和元数据。即使是没有状态操作的 stateless 查询（如简单的 select/filter），checkpoint 目录也会随着运行时间持续增长，因为每个 micro-batch 的 offset 信息都会被持久化。

但 stateless 查询的 checkpoint 增长速度通常很慢，每小时只增加几 KB 到几 MB。真正导致 OOM 的是 stateful 查询中的状态膨胀。stateful 操作包括 groupBy、agg、join、mapGroupsWithState 等需要跨 micro-batch 维护中间状态的算子。

stateful 状态膨胀的典型场景是：使用 groupBy 做聚合但没有设置 state TTL（生存时间）。每次 micro-batch 处理时，Spark 会将每个 key 的聚合结果存入状态中。如果 key 的基数持续增长（如按用户 ID 聚合且有新用户不断涌入），状态数据会无限增长，最终耗尽 executor 内存。

checkpoint 目录结构与损坏表现

checkpoint 目录的典型结构包含三个子目录：sources（保存数据源的 offset 信息）、offsets（每个 micro-batch 对应的 offset）、commits（已成功提交的 micro-batch 编号）。stateful 查询还会在 state 目录下保存状态数据。

checkpoint 损坏的常见表现是：重启查询时报 FileNotFoundException 或 InvalidObjectException，提示找不到某个 offset 文件或状态文件损坏。这通常发生在 executor 异常退出（如 OOM 被 kill）时，正在写入的 checkpoint 文件不完整。

另一个常见问题是 checkpoint 目录中出现大量小文件。每次 micro-batch 都会产生一批新文件，如果 micro-batch 间隔很短（如 1 秒），一天就会产生数万个文件。NameNode 需要管理这些文件的元数据，当文件数量超过百万级时，NameNode 的内存压力会显著增加，反过来拖慢整个 HDFS 集群的性能。

stateless 场景的排查与处理

如果查询是 stateless 的，checkpoint 增长慢，OOM 通常不是 checkpoint 的问题，而是 executor 内存配置不足或 GC 策略不当。检查 executor 的 spark.executor.memory 和 spark.executor.memoryOverhead 配置，后者建议设为 executor 内存的 10%-20%。

stateless 查询的 checkpoint 目录膨胀可以通过设置 checkpoint.cleaner.enabled=true（Spark 3.3+）来自动清理旧的 checkpoint 数据。对于更早版本，可以手动删除 offsets 和 commits 目录中的旧文件，但必须在查询停止的状态下操作。

stateful 场景的状态膨胀根因

stateful 查询的状态膨胀排查需要分三步走。第一步，确认是哪个算子的状态在增长。在 Spark UI 的 Structured Streaming 标签页中，查看各算子的 StateOperator 指标，找到 numRowsTotal 持续增长的算子。

第二步，分析 key 的基数增长模式。如果是窗口聚合（如 window 函数），检查窗口大小和水位线（watermark）设置。水位线决定了状态保留的时间窗口，如果水位线设置不合理或未设置，过期的状态不会被清理。为窗口聚合设置合理的 watermark 是控制状态增长的关键：withWatermark("event_time", "10 minutes") 表示超过 10 分钟的旧数据对应的状态会被清理。

第三步，检查是否有数据倾斜。某些 key 的数据量远大于其他 key，导致状态集中在少数几个分区上。解决方法是增加 shuffle 分区数（spark.sql.shuffle.partitions），让状态更均匀地分布在各分区。

三种恢复策略

当状态膨胀已经导致 OOM 或 checkpoint 损坏时，需要选择合适的恢复策略。三种策略适用于不同场景，选择错误可能导致数据重复或丢失。

策略一：checkpoint reset。删除整个 checkpoint 目录，重新启动查询。这是最简单粗暴的方法，适用于数据源支持重新读取的场景（如 Kafka 从 earliest offset 重新消费）。缺点是会重复处理历史数据，下游系统需要具备幂等写入能力。操作方法：停止查询 → 删除 checkpoint 目录 → 修改查询参数（如添加新的 outputMode）→ 重新启动。

策略二：full refresh（增量 offset 跳转）。不删除 checkpoint，而是修改 offset 文件跳转到当前最新的 offset。适用于数据源已经过期无法重新消费、但状态数据可以重建的场景。操作方法：停止查询 → 读取数据源当前最新的 offset → 手动修改 checkpoint 中 offsets 目录下的 offset 文件 → 重启查询。这种方法保留了原有的状态结构，但会跳过中间未处理的数据。

策略三：backup-backfill（备份回填）。先对当前状态做完整备份，然后清空 checkpoint 重启查询并从数据源的指定 offset 开始消费，同时将备份的状态数据通过 union 合并回新查询。这是最复杂但最安全的方法，适用于数据不能丢失也不能重复的关键业务场景。具体步骤：停止查询 → 备份状态数据到临时目录 → 清空 checkpoint → 启动新查询从指定 offset 开始 → 在查询逻辑中 union 备份的状态数据。

流批一体场景的 checkpoint 优化

在流批一体架构中，Spark Structured Streaming 的 checkpoint 管理需要与批处理任务协调。例如，当需要对历史数据做全量回刷时，不能简单地删除 checkpoint 重启，否则会丢失流处理的进度信息。

预防措施与最佳实践

预防状态膨胀比事后恢复更重要。以下几点建议值得在生产环境中落地：

始终设置 state TTL。对所有 stateful 查询使用 .option("statefulOperator.state.ttl", "1 hour") 或在聚合算子中使用 ttl 参数。TTL 的值根据业务需求设定，宁可设小后根据监控调整，也不要不设导致状态无限增长。

合理配置 micro-batch 间隔。默认的 ProcessingTime 触发模式下，间隔太短会导致 checkpoint 文件数量激增。对于状态量大的查询，建议将间隔设为 30 秒到 5 分钟，在延迟和 checkpoint 开销之间取得平衡。

定期监控 checkpoint 大小。在监控系统中添加 checkpoint 目录大小的趋势图，设置增长率告警。当 checkpoint 大小的增长速率突然加快时，说明状态正在快速膨胀，需要提前介入处理。

备份关键查询的 checkpoint。对于不能中断的核心业务查询，在查询稳定运行期间定期对 checkpoint 目录做快照备份。当需要做 full refresh 或版本升级时，可以先用备份的 checkpoint 恢复到已知的良好状态。