apache-spark - Spark 在运行期间用千兆字节的小文件填充临时目录

Question

我在 Linux 上的独立模式下使用带有 Spark 2.4 的 Pyspark 来聚合传入的数据并使用 将这些数据写入数据库。 Jupyter 笔记本 (目前用于测试)具有以下剥离内容:

spark = SparkSession.builder.appName("foo").getOrCreate()
df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:1234").option("subscribe", "bar”).load()
df2 = df.withWatermark("timestamp", "1 second").groupby(F.window('timestamp', "5 second")).agg(F.min("timestamp").alias('timestamp_window_min'))

def write_into_sink(df, epoch_id):
    df.write.jdbc(table="foo_agg", mode="append", [...])
    pass
query_write_sink = df2.writeStream.foreachBatch(write_into_sink).trigger(processingTime = "1 seconds").start()

在Spark的tmp目录下运行2小时后， tmp/temporary-[...]/state/0/目录下有几十个目录有很多小的 cec 和 delta 文件，加起来就是 6 GB 运行期间的磁盘空间。所以，我的问题是我无法运行脚本几个小时，因为磁盘已满；我怎么能运行它更长时间，比如几天甚至几个月？如果我关闭/杀死 python 内核，目录将被清除。

我已经关注了 Apache Spark does not delete temporary directories并调整 conf/spark-env.sh至 SPARK_WORKER_OPTS="spark.worker.cleanup.enabled=true"但在重新启动后仍然没有帮助，因为它只处理文件 Spark 运行后而不是期间。我也试过 SPARK_WORKER_OPTS="spark.worker.cleanup.enabled=true spark.worker.cleanup.appDataTtl=120"在具有相同不存在结果的同一个文件中。

那么，你知道如何摆脱spark 的tmp文件吗？运行期间 ?

Answer 1

您可以使用 cron 条目清理相关目录(也许，每 30 分钟删除一次超过 30 分钟的文件)，如下所示:

0/30 * * * * find /path_to_spark_directory/* -mmin +30 -exec rm -rf {} \;