如何在 Pyspark 中更改 fileoutputcommitter 算法的版本

Question

默认情况下，spark（2.4.4）使用 MapReduce.fileoutputcommitter.algorithm.version 1。我正在尝试将其更改为版本 2。 spark-UI 并sparkCtx._conf.getAll()显示版本 2，但 pyspark 仍然使用版本 1（临时文件夹）将数据写入 S3正在创建）。我已经尝试过这些东西。spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version   2在 spark-defaults.conf 中设置为 spark 会话定义 config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version","2")。df.write 中的设置选项如下： df.write.option("mapreduce.fileoutputcommitter.algorithm.version", "2")

Answer 1

但 pyspark 仍然使用版本 1 将数据写入 S3（正在创建临时文件夹）。

首先，v1 和 v2 算法都使用临时文件。如MAPREDUCE-6336中所述

算法版本 2 更改了 commitTask、recoverTask 和 commitJob 的行为。

1. commitTask 将 $joboutput/_temporary/$appAttemptID/_temporary/$taskAttemptID/ 中的所有文件重命名为 $joboutput/

2. recoverTask 严格来说是一个 nop，但是对于从版本 1 升级到版本 2 的情况，它会检查 $joboutput/_temporary/($appAttemptID - 1)/$taskID/ 中是否有任何文件并将它们重命名为 $joboutput/

3. commitJob 删除 $joboutput/_temporary 并写入 $joboutput/_SUCCESS

因此，请确保您实际看到的是与 v1 而不是 v2 对应的更改。

此外spark.hadoop，选项适用于上下文而不是特定的写入，因此您的第三次尝试根本不起作用。

其余的应该是等价的（第二个，如果之前设置过SparkContext）。