現象

大家在使用 Apache Spark 2.x 的時候可能會遇到這種現象：雖然我們的 Spark Jobs 已經全部完成了，但是我們的程序卻還在執行。比如我們使用 Spark SQL 去執行一些 SQL，這個 SQL 在最后生成了大量的文件。然后我們可以看到，這個 SQL 所有的 Spark Jobs 其實已經運行完成了，但是這個查詢語句還在運行。通過日志，我們可以看到 driver 節點正在一個一個地將 tasks 生成的文件移動到最終表的目錄下面，當我們作業生成的文件很多的情況下，就很容易產生這種現象。本文將給大家介紹一種方法來解決這個問題。

為什么會造成這個現象

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

問題就出在了 Hadoop 2.x 的 FileOutputCommitter 實現FileOutputCommitter 里面有兩個值得注意的方法：commitTask 和 commitJob。在 Hadoop 2.x 的FileOutputCommitter 實現里面，mapreduce.fileoutputcommitter.algorithm.version 參數控制著 commitTask 和 commitJob 的工作方式。具體代碼如下（為了說明方便，我去掉了無關緊要的語句，完整代碼可以參見 FileOutputCommitter.java）：

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

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

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