Spark结构化流检查点大小巨大

Posted 2023-04-15

【中文标题】Spark结构化流检查点大小巨大

【英文标题】：Spark structured streaming checkpoint size huge

【发布时间】：2019-12-05 17:38:48

【问题描述】：

我正在从事 Spark 结构化流式处理项目，我在 chechpoint 中遇到了一个问题。

在我们的 hdfs 中，我们有 25 天的保留政策及其每日分区，我们将每天从 hdfs 中删除文件，但在我的 spark 流中，我的 checkpnt 文件会保存工作开始时的所有文件名，但如果我清理我的checkpnt 目录我需要重新开始我的工作 25 天，所以我需要根据我的保留策略删除我的 checkpnt 文件，但 checkpnt 中最新的 .compact 文件存储了所有从开始的文件名，请帮助我解决这个问题。

【问题讨论】：

这是结构化流（可能）还是火花流？您能显示流式查询的“写入路径”吗？您可以将它与目录列表一起包含吗？谢谢。

这是火花结构化流媒体

【参考方案1】：

您不应手动删除检查点文件夹。在 spark 配置中有一个连接器： https://spark.apache.org/docs/latest/configuration.html#memory-management

spark.cleaner.referenceTracking.cleanCheckpoints

对于 DStreams 还有一个cleanup 方法：

https://github.com/apache/spark/blob/master/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStreamCheckpointData.scala

【讨论】：

如果文件名超过 25 天，它将根据什么进行引用和识别，以及我在哪里可以设置此属性？

如果我们在上面设置属性=true，它将控制是否在引用超出范围时清理检查点文件，这里我的问题是系统如何显示超出范围的范围？

这适用于 Spark Core（用于 RDD 检查点）和旧模块 Spark Streaming（不是结构化流）。

以上属性不适用于我的情况，它仍然保留旧文件，并且由于我们的保留而全部从 hdfs 中删除，但文件名仍然存在于检查点文件中

据我所知，主要问题不是尺寸太大，而是保留策略会在 25 天后删除检查点文件夹。我不知道你的流应用程序是有状态的还是无状态的，但如果你只是想有一个恢复选项，你可以在 ZooKeeper 或 HBase 外部存储 Kafka 偏移量：blog.cloudera.com/…。如果它是一个有状态的应用程序，出于逻辑原因您需要检查点，因此您可能需要修改您的保留策略。