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

Question

默认情况下，spark（2.4.4）使用MapReduce.fileoutputcommitter.algorithm.version 1。我正在尝试将其更改为版本2。spark UI和sparkCtx._conf.getAll()显示了版本2，但pyspark仍然使用版本1在S3中写入数据（正在创建临时文件夹）。我试过这些东西

1. 在spark-defaults.conf中设置spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2
2. 为spark会话定义配置（“spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version”，“2”）
3. 在df.write中设置选项，如下所示： df.write.option("mapreduce.fileoutputcommitter.algorithm.version", "2")

Answer 1

but pyspark still writes the data in S3 using version 1(temporary folders are creating).

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

Algorithm version 2 changes the behavior of commitTask, recoverTask, and commitJob.

1. commitTask renames all files in $joboutput/_temporary/$appAttemptID/_temporary/$taskAttemptID/ to $joboutput/

2. recoverTask is a nop strictly speaking, but for upgrade from version 1 to version 2 case, it checks if there are any files in $joboutput/_temporary/($appAttemptID - 1)/$taskID/ and renames them to $joboutput/

3. commitJob deletes $joboutput/_temporary and writes $joboutput/_SUCCESS

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

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

其余的应该是等效的（第二个，如果在SparkContext启动之前设置）