如何清理Spark结构化流Checkpoint State历史批次数据并阻止Snapshot增长？

作业背景与代码 #

我有一个Spark结构化流作业，从Kafka拉取数据，通过forEachBatch将数据写入Neo4j，核心代码如下：

StreamingQuery query = eventsDf
        .writeStream()
        .queryName("streaming")
        .outputMode("update")
        .trigger(Trigger.ProcessingTime(80000))
        .foreachBatch(
                (VoidFunction2<Dataset<Row>, Long>) (dfBatch, batchId) -> {
                    // 写入Neo4j的逻辑
                }
        )
        .option("checkpointLocation", "src/main/resources/checkpoint")
        .start();

其中eventsDf包含经过聚合和过滤的数据，无时间戳列，无法使用窗口/水印机制。

问题场景与需求 #

• 第一个批次会生成正常数据，但80秒后的第二个批次因使用max聚合函数，结果中包含前一批次的数据；
• 需求：
  1. 丢弃前一批次的数据，只保留当前批次的聚合结果；
  2. 解决Checkpoint目录下State文件夹（包含delta和checkpoint文件）持续扩容的问题，即使配置了minDeltasForSnapshot，文件仍在不断增长。

已尝试的Spark配置 #

"spark.sql.streaming.minBatchesToRetain", 2
"spark.sql.streaming.stateStore.minDeltasForSnapshot", 2
"cleanSource", "delete"
"spark.sql.streaming.forceDeleteTempCheckpointLocation", true

更新后的情况 #

我在代码中添加了batchId列到groupby的分组键中（先通过.withColumn()添加batchId，再执行分组和聚合），修改后delta文件不再持续增长，但snapshot文件仍在不断增长，请问如何阻止该情况？

解决方案 #

一、解决跨批次重复聚合数据问题

通过将batchId纳入分组键，让每个批次的聚合结果完全独立，是正确的思路。这样每个批次的聚合仅基于当前批次的数据，不会和历史批次的状态混淆，自然就不会包含前一批次的结果。示例代码如下：

import org.apache.spark.sql.functions;

// 聚合前添加当前批次ID列（Spark 3.2+支持current_batch_id()，低版本可通过foreachBatch的batchId参数传入）
Dataset<Row> eventsWithBatchId = eventsDf
    .withColumn("batch_id", functions.current_batch_id());

// 分组时将batch_id加入分组键，确保每个批次的聚合独立
Dataset<Row> aggregatedDf = eventsWithBatchId
    .groupBy("your_group_key", "batch_id") // 替换为实际业务分组键
    .agg(functions.max("target_column").alias("max_value")); // 你的max聚合操作

后续在forEachBatch中处理的dfBatch就是当前批次的独立聚合结果，直接写入Neo4j即可，无需额外过滤。

二、解决State目录snapshot文件持续增长问题

snapshot文件是状态存储的快照，默认Spark会保留10个快照，这是导致其持续增长的核心原因。可以通过以下配置调整：

1. 限制快照保留数量
   设置spark.sql.streaming.stateStore.numSnapshotsRetained参数，指定仅保留最近的1-2个快照：

   spark.sql.streaming.stateStore.numSnapshotsRetained=1
   

   该参数控制保留的快照文件数量，降低后可直接减少快照占用的存储空间。

2. 配合批次保留策略强化清理
   结合spark.sql.streaming.minBatchesToRetain参数，确保旧批次的状态被及时清理：

   spark.sql.streaming.minBatchesToRetain=1
   

   该参数控制Spark保留的最小批次数量，设为1时，Spark会在处理完下一个批次后自动清理上一批次的状态数据，进一步减少State目录的占用。

3. 应急手动清理（非运行时）
   如果配置生效需要时间，可在作业停止时手动删除旧的snapshot文件，但注意保留最近的1-2个快照，避免作业重启时无法恢复状态。不建议在作业运行时手动操作。

内容的提问来源于stack exchange，提问作者Rahul Shah