Spark 数据帧检查点清理

Question

我在 spark 中有一个数据帧，其中已加载来自 Hive 的整个分区，我需要在对数据进行一些修改后打破沿袭以覆盖相同的分区。但是，当火花作业完成时，我留下了来自 HDFS 上检查点的数据。为什么 Spark 不自行清理它或者我缺少什么？

spark.sparkContext.setCheckpointDir("/home/user/checkpoint/")
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

val df = spark.table("db.my_table").filter(col("partition").equal(2))

// ... transformations to the dataframe

val checkpointDf = df.checkpoint()
checkpointDf.write.format("parquet").mode(SaveMode.Overwrite).insertInto("db.my_table")

在此之后，我在 HDFS 上有了这个文件：

/home/user/checkpoint/214797f2-ce2e-4962-973d-8f215e5d5dd8/rdd-23/part-00000

每次我运行 spark 作业时，我都会得到一个新目录，其中包含一个新的唯一 ID，其中包含数据帧中每个 RDD 的文件。

Answer 1

Spark 具有用于检查点文件清理的隐式机制。

在 spark-defaults.conf 中添加此属性。

spark.cleaner.referenceTracking.cleanCheckpoints  true #Default is false

您可以在 Spark official configuration page中找到更多关于 Spark 配置的信息

如果您想从 HDFS 中删除检查点目录，可以使用 Python 将其删除，在脚本末尾您可以使用此命令 rmtree。

此属性spark.cleaner.referenceTracking.cleanCheckpoints 与true 一样，允许清理程序删除检查点目录中的旧检查点文件。