首頁猿問如何在 Pyspark 中更改...

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

Python

拉風(fēng)的咖菲貓 2022-10-18 16:10:11

默認(rèn)情況下，spark（2.4.4）使用 MapReduce.fileoutputcommitter.algorithm.version 1。我正在嘗試將其更改為版本 2。 spark-UI 并sparkCtx._conf.getAll()顯示版本 2，但 pyspark 仍然使用版本 1（臨時(shí)文件夾）將數(shù)據(jù)寫入 S3正在創(chuàng)建）。我已經(jīng)嘗試過這些東西。spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2在 spark-defaults.conf 中設(shè)置為 spark 會(huì)話定義 config("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version","2")。df.write 中的設(shè)置選項(xiàng)如下： df.write.option("mapreduce.fileoutputcommitter.algorithm.version", "2")

查看完整描述

1 回答

慕田峪9158850

TA貢獻(xiàn)1794條經(jīng)驗(yàn) 獲得超7個(gè)贊

但 pyspark 仍然使用版本 1 將數(shù)據(jù)寫入 S3（正在創(chuàng)建臨時(shí)文件夾）。

首先，v1 和 v2 算法都使用臨時(shí)文件。如MAPREDUCE-6336中所述

算法版本 2 更改了 commitTask、recoverTask 和 commitJob 的行為。
commitTask 將 $joboutput/_temporary/$appAttemptID/_temporary/$taskAttemptID/ 中的所有文件重命名為 $joboutput/
recoverTask 嚴(yán)格來說是一個(gè) nop，但是對(duì)于從版本 1 升級(jí)到版本 2 的情況，它會(huì)檢查 $joboutput/_temporary/($appAttemptID - 1)/$taskID/ 中是否有任何文件并將它們重命名為 $joboutput/
commitJob 刪除 $joboutput/_temporary 并寫入 $joboutput/_SUCCESS

因此，請(qǐng)確保您實(shí)際看到的是與 v1 而不是 v2 對(duì)應(yīng)的更改。

此外spark.hadoop，選項(xiàng)適用于上下文而不是特定的寫入，因此您的第三次嘗試根本不起作用。

其余的應(yīng)該是等價(jià)的（第二個(gè)，如果之前設(shè)置過SparkContext）。

反對(duì) 回復(fù) 2022-10-18