Apache Spark 2.0在作業(yè)完成時(shí)卻花費很長(cháng)時(shí)間結束怎

這篇文章給大家分享的是有關(guān)Apache Spark 2.0在作業(yè)完成時(shí)卻花費很長(cháng)時(shí)間結束怎么辦的內容。小編覺(jué)得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過(guò)來(lái)看看吧。

現象

大家在使用 Apache Spark 2.x 的時(shí)候可能會(huì )遇到這種現象：雖然我們的 Spark Jobs 已經(jīng)全部完成了，但是我們的程序卻還在執行。比如我們使用 Spark SQL 去執行一些 SQL，這個(gè) SQL 在最后生成了大量的文件。然后我們可以看到，這個(gè) SQL 所有的 Spark Jobs 其實(shí)已經(jīng)運行完成了，但是這個(gè)查詢(xún)語(yǔ)句還在運行。通過(guò)日志，我們可以看到 driver 節點(diǎn)正在一個(gè)一個(gè)地將 tasks 生成的文件移動(dòng)到最終表的目錄下面，當我們作業(yè)生成的文件很多的情況下，就很容易產(chǎn)生這種現象。

為什么會(huì )造成這個(gè)現象

Spark 2.x 用到了 Hadoop 2.x，其將生成的文件保存到 HDFS 的時(shí)候，最后會(huì )調用了 saveAsHadoopFile，而這個(gè)函數在里面用到了 FileOutputCommitter，如下：

問(wèn)題就出在了 Hadoop 2.x 的 FileOutputCommitter 實(shí)現FileOutputCommitter 里面有兩個(gè)值得注意的方法：commitTask 和 commitJob。在 Hadoop 2.x 的FileOutputCommitter 實(shí)現里面，mapreduce.fileoutputcommitter.algorithm.version 參數控制著(zhù) commitTask 和 commitJob 的工作方式。具體代碼如下（為了說(shuō)明方便，我去掉了無(wú)關(guān)緊要的語(yǔ)句，完整代碼可以參見(jiàn) FileOutputCommitter.java）：

大家可以看到 commitTask 方法里面，有個(gè)條件判斷 algorithmVersion == 1，這個(gè)就是 mapreduce.fileoutputcommitter.algorithm.version 參數的值，默認為1；如果這個(gè)參數為1，那么在 Task 完成的時(shí)候，是將 Task 臨時(shí)生成的數據移到 task 的對應目錄下，然后再在 commitJob 的時(shí)候移到最終作業(yè)輸出目錄，而這個(gè)參數,在 Hadoop 2.x 的默認值就是 1！這也就是為什么我們看到 job 完成了，但是程序還在移動(dòng)數據，從而導致整個(gè)作業(yè)尚未完成，而且最后是由 Spark 的 Driver 執行 commitJob 函數的，所以執行的慢也是有到底的。

而我們可以看到，如果我們將 mapreduce.fileoutputcommitter.algorithm.version 參數的值設置為 2，那么在 commitTask 執行的時(shí)候，就會(huì )調用 mergePaths 方法直接將 Task 生成的數據從 Task 臨時(shí)目錄移動(dòng)到程序最后生成目錄。而在執行 commitJob 的時(shí)候，直接就不用移動(dòng)數據了，自然會(huì )比默認的值要快很多。

注意，其實(shí)在 Hadoop 2.7.0 之前版本，我們可以將 mapreduce.fileoutputcommitter.algorithm.version 參數設置為非1的值就可以實(shí)現這個(gè)目的，因為程序里面并沒(méi)有限制這個(gè)值一定為2,。不過(guò)到了 Hadoop 2.7.0，mapreduce.fileoutputcommitter.algorithm.version 參數的值必須為1或2，具體參見(jiàn) MAPREDUCE-4815。

怎么在 Spark 里面設置這個(gè)參數

問(wèn)題已經(jīng)找到了，我們可以在程序里面解決這個(gè)問(wèn)題。有以下幾種方法：

• 直接在 conf/spark-defaults.conf 里面設置 spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2，這個(gè)是全局影響的。

• 直接在 Spark 程序里面設置，spark.conf.set("mapreduce.fileoutputcommitter.algorithm.version", "2")，這個(gè)是作業(yè)級別的。

• 如果你是使用 Dataset API 寫(xiě)數據到 HDFS，那么你可以這么設置 dataset.write.option("mapreduce.fileoutputcommitter.algorithm.version", "2")。

不過(guò)如果你的 Hadoop 版本為 3.x，mapreduce.fileoutputcommitter.algorithm.version 參數的默認值已經(jīng)設置為2了，具體參見(jiàn) MAPREDUCE-6336 和 MAPREDUCE-6406。

因為這個(gè)參數對性能有一些影響，所以到了 Spark 2.2.0，這個(gè)參數已經(jīng)記錄在 Spark 配置文檔里面了 configuration.html，具體參見(jiàn) SPARK-20107。