[轉] Hadoop MapReduce開發最佳實踐（上篇）

前言

本文是Hadoop最佳實踐系列第二篇，上一篇爲《Hadoop管理員的十個最佳實踐》。

MapRuduce開發對於大多數程序員都會以爲略顯複雜，運行一個WordCount（Hadoop中hello word程序）不只要熟悉MapRuduce模型，還要了解Linux命令（儘管有Cygwin，但在Windows下運行MapRuduce仍然很麻煩），此外還要學習程序的打包、部署、提交job、調試等技能，這足以讓不少學習者望而退步。

因此如何提升MapReduce開發效率便成了你們很關注的問題。但Hadoop的Committer早已經考慮到這些問題，從而開發了ToolRunner、MRunit（MapReduce最佳實踐第二篇中會介紹）、MiniMRCluster、MiniDFSCluster等輔助工具，幫助解決開發、部署等問題。舉一個本身親身的例子：

 

某週一和搭檔(結對編程)決定重構一個完成近10項統計工做的MapRuduce程序，這個MapReduce（從Spring項目移植過來的）,由於依賴Spring框架(原生Spring，非Spring Hadoop框架),致使性能難以忍受，咱們決定將Spring從程序中剔除。重構以前程序運行是正確的，因此咱們要保障重構後運行結果與重構前一致。搭檔說，爲何咱們不用TDD來完成這個事情呢？因而咱們研究並應用了MRunit，使人意想不到的是，重構工做只用了一天就完成，剩下一天咱們進行用findbug掃描了代碼，進行了集成測試。此次重構工做咱們沒有給程序帶來任何錯誤，不但如此咱們還擁有了可靠的測試和更加穩固的代碼。這件事情讓咱們很爽的同時，也在思考關於MapReduce開發效率的問題，要知道此次重構咱們以前評估的時間是一週，我把這個事情分享到EasyHadoop羣裏，你們頗有興趣，一個朋友問到，大家的評估太不許確了，爲何開始不評估2天完成呢？我說若是咱們沒有使用MRUnit，真的是須要一週才能完成。由於有它單元測試，我能夠在5秒內獲得我本次修改的反饋，不然至少須要10分鐘（編譯、打包、部署、提交MapReduce、人工驗證結果正確性），並且重構是個反覆修改，反覆運行，獲得反饋，再修改、再運行、再反饋的過程，MRunit在這裏幫了大忙。

相同智商、相同工做經驗的開發人員，藉助有效的工具和方法，居然能夠帶來如此大的開發效率差距，不得不讓人驚詫！

PS. 本文基於Hadoop 1.0（Cloudera CDH3uX）。本文適合讀者：Hadoop初級、中級開發者。

 

1. 使用ToolRunner讓參數傳遞更簡單

關於MapReduce運行和參數配置，你是否有下面的煩惱：

1. 將MapReduce Job配置參數寫到java代碼裏，一旦變動意味着修改java文件源碼、編譯、打包、部署一連串事情。
2. 當MapReduce 依賴配置文件的時候，你須要手工編寫java代碼使用DistributedCache將其上傳到HDFS中，以便map和reduce函數能夠讀取。
3. 當你的map或reduce 函數依賴第三方jar文件時，你在命令行中使用」-libjars」參數指定依賴jar包時，但根本沒生效。

其實，Hadoop有個ToolRunner類，它是個好東西，簡單好用。不管在《Hadoop權威指南》仍是Hadoop項目源碼自帶的example，都推薦使用ToolRunner。

下面咱們看下src/example目錄下WordCount.java文件，它的代碼結構是這樣的：

public class WordCount {
    // 略...
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, 
                                            args).getRemainingArgs();
        // 略...
        Job job = new Job(conf, "word count");
        // 略...
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

WordCount.java中使用到了GenericOptionsParser這個類，它的做用是將命令行中參數自動設置到變量conf中。舉個例子，好比我但願經過命令行設置reduce task數量，就這麼寫：

bin/hadoop jar MyJob.jar com.xxx.MyJobDriver -Dmapred.reduce.tasks=5

上面這樣就能夠了，不須要將其硬編碼到java代碼中，很輕鬆就能夠將參數與代碼分離開。

其它經常使用的參數還有」-libjars」和-「files」，使用方法一塊兒送上：

bin/hadoop jar MyJob.jar com.xxx.MyJobDriver -Dmapred.reduce.tasks=5 \ 
    -files ./dict.conf  \
    -libjars lib/commons-beanutils-1.8.3.jar,lib/commons-digester-2.1.jar

參數」-libjars」的做用是上傳本地jar包到HDFS中MapReduce臨時目錄並將其設置到map和reduce task的classpath中；參數」-files」的做用是上傳指定文件到HDFS中mapreduce臨時目錄，並容許map和reduce task讀取到它。這兩個配置參數其實都是經過DistributeCache來實現的。

至此，咱們尚未說到ToolRunner，上面的代碼咱們使用了GenericOptionsParser幫咱們解析命令行參數，編寫ToolRunner的程序員更懶，它將 GenericOptionsParser調用隱藏到自身run方法，被自動執行了，修改後的代碼變成了這樣：

public class WordCount extends Configured implements Tool {

    @Override
    public int run(String[] arg0) throws Exception {
        Job job = new Job(getConf(), "word count");
        // 略...
        System.exit(job.waitForCompletion(true) ? 0 : 1);
        return 0;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new WordCount(), args);
        System.exit(res);
    }
}

看看代碼上有什麼不一樣：

1. 讓WordCount繼承Configured並實現Tool接口。
2. 重寫Tool接口的run方法，run方法不是static類型，這很好。
3. 在WordCount中咱們將經過getConf()獲取Configuration對象。

關於GenericOptionsParser更多用法，請點擊這裏：GenericOptionsParser.html

推薦指數：★★★★

推薦理由：經過簡單的幾步，就能夠實現代碼與配置隔離、上傳文件到DistributeCache等功能。修改MapReduce參數不須要修改java代碼、打包、部署，提升工做效率。

2. 有效使用Hadoop源碼

做爲MapReduce程序員不可避免的要使用Hadoop源碼，Why？記得2010剛接觸hadoop的時候，老是搞不清舊api和新api的使用方法。寫了一段程序，在一個新api裏面調用某個方法每次都是返回Null，很是惱火，後來附上源碼發現，這個方法真的就是隻作了「return null」並無給予實現，最後只得想其它方法曲線救國。總之要想真正瞭解MapReduce開發，源碼是不可缺乏的工具。

下面是個人源碼使用實踐，步驟有點麻煩不過配置一次就好：

1. Eclipse中建立Hadoop源碼項目

1.1 下載並解壓縮Hadoop分發包（一般是tar.gz包）

1.2 Eclipse中新建Java項目

1.3 將解壓後hadoop源碼包/src目錄中core, hdfs, mapred, tool幾個目錄（其它幾個源碼根據須要進行選擇）copy到eclipse新建項目的src目錄。

1.4 右鍵點擊eclipse項目，選擇「Properties」，在彈出對話框中左邊菜單選擇「Java Build Path」：
    a) 點擊「Source」標籤。先刪除src這個目錄，而後依次添加剛纔copy過來的目錄
    b) 點擊當前對話框「Libaries」，點擊「Add External JARs」，在彈出窗口中添加$HADOOPHOME下幾個hadoop程序jar包，而後再次添加$HADOOPHOME /lib、$HADOOP_HOME /lib/jsp-2.1兩個目錄下全部jar包，最後還要添加ANT項目lib目錄下ant.jar文件。

1.5 此時源碼項目應該只有關於找不到sun.security包的錯誤了。這時咱們仍是在「Libraries」這個標籤中，展開jar包列表最低下的「JRE System Library」，雙擊」Access rules」，在彈出窗口中點擊「add按鈕」，而後在新對話框中"Resolution"下拉框選擇"Accessible"，"Rule Pattern"填寫*/，保存後就OK了。以下圖：

2. 如何使用這個源碼項目呢？

好比我知道Hadoop某個源碼文件的名稱，在eclipse中能夠經過快捷鍵「Ctrl + Shift + R」調出查找窗口，輸入文件名，如「MapTask」，那能夠打開這個類的源碼了。

還有個使用場景，當咱們編寫MapReduce程序的時候，我想直接打開某個類的源碼，經過上面的操做仍是有點麻煩，好比我想看看Job類是如何實現的，當我點擊它的時候會出現下面的情景：

解決辦法很簡單：

點擊圖中「Attach Source」按鈕-> 點擊「Workspace」按鈕->選擇剛纔新建的Hadoop源碼項目。完成後源碼應該就蹦出來了。

總結一下，本實踐中咱們得到了什麼功能：

1. 知道hadoop源碼文件名，快速找到該文件
2. 寫程序的時候直接查看Hadoop相關類源碼
3. Debug程序的時候，能夠直接進入源碼查看並跟蹤運行

推薦指數：★★★★

推薦理由：經過源碼能夠幫助咱們更深刻了解Hadoop，能夠幫助咱們解決複雜問題

3. 正確使用壓縮算法

下表資料引用cloudera官方網站的一篇博客，原文點這裏。

Compression	File	Size(GB)	Compression Time (s)	Decompression Time (s)
None	some_logs	8.0	-	-
Gzip	some_logs.gz	1.3	241	72
LZO	some_logs.lzo	2.0	55	35

上面表格與筆者集羣實際環境測試結果一致，因此咱們能夠得出以下結論：

1. LZO文件的壓縮和解壓縮性能要遠遠好於Gzip文件。
2. 相同文本文件，使用Gzip壓縮能夠比LZO壓縮大幅減小磁盤空間。

上面的結論對咱們有什麼幫助呢？在合適的環節使用合適壓縮算法。

在中國的帶寬成本是很是貴的，費用上要遠遠高於美國、韓國等國家。因此在數據傳輸環節，咱們但願使用了Gzip算法壓縮文件，目的是減小文件傳輸量，下降帶寬成本。使用LZO文件做爲MapReduce文件的輸入（建立lzo index後是支持自動分片輸入的）。對於大文件，一個map task的輸入將變爲一個block，而不是像Gzip文件同樣讀取整個文件，這將大幅提高MapReduce運行效率。

主流傳輸工具FlumeNG和scribe默認都是非壓縮傳輸的（都是經過一行日誌一個event進行控制的），這點你們在使用時要注意。FlumeNG能夠自定義組件方式實現一次傳輸多條壓縮數據，而後接收端解壓縮的方式來實現數據壓縮傳輸，scribe沒有使用過不評論。

另外值得一提的就是snappy，它是由Google開發並開源的壓縮算法的，是Cloudera官方大力提倡在MapReduce中使用的壓縮算法。它的特色是：與LZO文件相近的壓縮率的狀況下，還能夠大幅提高壓縮和解壓縮性能，可是它做爲MapReduce輸入是不能夠分割的。

延伸內容：

Cloudera官方Blog對Snappy介紹：

http://blog.cloudera.com/blog/2011/09/snappy-and-hadoop/

老外上傳的壓縮算法性能測試數據：

http://pastebin.com/SFaNzRuf

推薦指數：★★★★★

推薦理由：壓縮率和壓縮性能必定程度是矛盾體，如何均衡取決於應用場景。使用合適壓縮算法直接關係到老闆的錢，若是可以節省成本，體現程序員的價值。

4. 在合適的時候使用Combiner

map和 reduce 函數的輸入輸出都是key-value，Combiner和它們是同樣的。做爲map和reduce的中間環節，它的做用是聚合map task的磁盤，減小map端磁盤寫入，減小reduce端處理的數據量，對於有大量shuffle的job來講，性能每每取決於reduce端。由於reduce 端要通過從map端copy數據、reduce端歸併排序，最後纔是執行reduce方法，此時若是能夠減小map task輸出將對整個job帶來很是大的影響。

何時可使用Combiner？

好比你的Job是WordCount，那麼徹底能夠經過Combiner對map 函數輸出數據先進行聚合，而後再將Combiner輸出的結果發送到reduce端。

何時不能使用Combiner？

WordCount在reduce端作的是加法，若是咱們reduce需求是計算一大堆數字的平均數，則要求reduce獲取到所有的數字進行計算，才能夠獲得正確值。此時，是不能使用Combiner的，由於會其會影響最終結果。 注意事項：即便設置Combiner，它也不必定被執行（受參數min.num.spills.for.combine影響），因此使用Combiner的場景應保證即便沒有Combiner，咱們的MapReduce也能正常運行。

推薦指數：★★★★★

推薦理由：在合適的場景使用Combiner，能夠大幅提高MapReduce 性能。

5. 經過回調通知知道MapReduce何時完成

你知道何時MapReduce完成嗎？知道它執行成功或是失敗嗎？

Hadoop包含job通知這個功能，要使用它很是容易，藉助咱們實踐一的ToolRunner，在命令行裏面就能夠進行設置，下面是一個例子：

hadoop jar MyJob.jar com.xxx.MyJobDriver \
-Djob.end.notification.url=http://moniter/mapred_notify/\$jobId/\$jobStatus 

經過上面的參數設置後，當MapReduce完成後將會回調我參數中的接口。其中$jobId和$jobStatus會自動被實際值代替。

上面在$jobId和$jobStatus兩個變量前，我添加了shell中的轉義符」\」，若是使用java代碼設置該參數是不須要轉義符的。

總結下：看看咱們經過該實踐能夠得到什麼？

1. 獲取MapReduce運行時間和回調完成時間，能夠分析最耗時Job，最快完成Job。
2. 經過MapReduce運行狀態（包括成功、失敗、Kill），能夠第一時間發現錯誤，並通知運維。
3. 經過獲取MapReduce完成時間，能夠第一時間經過用戶，數據已經計算完成，提高用戶體驗

Hadoop這塊功能的源碼文件是JobEndNotifier.java，能夠立刻經過本文實踐二看看究竟。其中下面兩個參數就是我經過翻源碼的時候發現的，若是但願使用該實踐趕忙經過ToolRunner設置上吧（別忘了加-D，格式是-Dkey=value）。

1. job.end.retry.attempts // 設置回調通知retry次數
2. job.end.retry.interval // 設置回調時間間隔，單位毫秒

固然若是hadoop沒有提供Job狀態通知的功能，咱們也能夠經過採用阻塞模式提交MapReduce Job，而後Job完成後也能夠獲知其狀態和運行時間。

推薦指數：★★★

推薦理由：對mapreduce job監控最省事有效的辦法，沒有之一。