火花结构化流 - 检查点元数据无限期地生长

Question

我使用Spark Struture流媒体3.1.2。我需要使用S3来存储检查点元数据（我知道，它不是检查点元数据的最佳存储）。压实间隔为10（默认），我设置spark.sql.sql.streaming.minbatchestoretain = 5。当作业运行几周后，检查点时间大大增加（导致加工时Dalay几分钟）。我看了检查点元数据结构。那里有一条繁重的路径：检查点/源/0。单个.compact文件权重25GB。我研究了它的内容，并且包含自批次0以来的所有条目（当前批次约为25000）。

我尝试了一些参数来从紧凑文件中删除已经处理过的数据，即： spark.cleaner.referencetracking.cleancheckpoints = true - 不起作用。正如我在代码中看到的那样，它与以前版本的流媒体相关，不是吗？ spark.sql.streaming.filesource.log.deletion = true and spark.sql.Streaming.filesink.log.deletion = true不起作用。

即使处理了所有数据（除了最近的检查点除外），紧凑型文件存储了完整的历史记录，因此我希望大多数条目都会被删除。是否有任何参数可以从紧凑文件中删除条目或不时优雅地删除紧凑型文件？

现在，我正在测试停止作业时，正在测试场景，删除大多数checkpoint/source/0/*文件，仅保留一些最近的检查点（未压实），然后重新运行工作。该作业从最近的检查点正确恢复。当涉及到检查点的压实时，它会失败，而缺少最近的压实文件。我可能需要编辑最新的紧凑型文件（而不是删除它），只保留少数最近的记录。看起来可能会解决我的问题，但是使用手动删除检查点文件的情况看起来很丑，因此我更喜欢Spark管理的东西。

Answer 1

对于后代：问题是fileStreamSourcelog类。我需要覆盖方法shordretain，默认情况下返回true，其文档说：

默认实现保留所有日志条目。实现应覆盖改变行为的方法。

Answer 2

请参阅 。主要的option（“保留”，保留）解决了同样的问题。