Hadoop 實戰實例

       Hadoop 是Google MapReduce的一個Java實現。MapReduce是一種簡化的分布式編程模式，讓程序自動分布到一個由普通機器組成的超大集群上并發執行。就如同java程序員可以不考慮內存泄露一樣， MapReduce的run-time系統會解決輸入數據的分布細節，跨越機器集群的程序執行調度，處理機器的失效，并且管理機器之間的通訊請求。這樣的模式允許程序員可以不需要有什么并發處理或者分布式系統的經驗，就可以處理超大的分布式系統得資源。

一、概論

    作為Hadoop程序員，他要做的事情就是：
    1、定義Mapper，處理輸入的Key-Value對，輸出中間結果。
    2、定義Reducer，可選，對中間結果進行規約，輸出最終結果。
    3、定義InputFormat 和OutputFormat，可選，InputFormat將每行輸入文件的內容轉換為Java類供Mapper函數使用，不定義時默認為String。
    4、定義main函數，在里面定義一個Job并運行它。
    

    然后的事情就交給系統了。
    1.基本概念：Hadoop的HDFS實現了google的GFS文件系統，NameNode作為文件系統的負責調度運行在 master，DataNode運行在每個機器上。同時Hadoop實現了Google的MapReduce，JobTracker作為 MapReduce的總調度運行在master，TaskTracker則運行在每個機器上執行Task。

    2.main()函數，創建JobConf，定義Mapper，Reducer，Input/OutputFormat 和輸入輸出文件目錄，最后把Job提交給JobTracker，等待Job結束。

    3.JobTracker，創建一個InputFormat的實例，調用它的getSplits()方法，把輸入目錄的文件拆分成FileSplist作為Mapper task 的輸入，生成Mapper task加入Queue。

    4.TaskTracker 向 JobTracker索求下一個Map/Reduce。
      
     Mapper Task先從InputFormat創建RecordReader，循環讀入FileSplits的內容生成Key與Value，傳給Mapper函數，處理完后中間結果寫成SequenceFile.
     Reducer Task 從運行Mapper的TaskTracker的Jetty上使用http協議獲取所需的中間內容（33%），Sort/Merge后（66%），執行Reducer函數，最后按照OutputFormat寫入結果目錄。 

      TaskTracker 每10秒向JobTracker報告一次運行情況，每完成一個Task10秒后，就會向JobTracker索求下一個Task。

      Nutch項目的全部數據處理都構建在Hadoop之上，詳見Scalable Computing with Hadoop。

二、程序員編寫的代碼

 （可以查看hadoop-examples-0.20.203.0.jar,里面也有一個類grep）

    我們做一個簡單的分布式的Grep，簡單對輸入文件進行逐行的正則匹配，如果符合就將該行打印到輸出文件。因為是簡單的全部輸出，所以我們只要寫Mapper函數，不用寫Reducer函數，也不用定義Input/Output Format。

package  demo.hadoop
public   class  HadoopGrep {
     public   static   class  RegMapper  extends  MapReduceBase  implements  Mapper {
               private  Pattern pattern;
               public   void  configure(JobConf job) {
                         pattern  =  Pattern.compile(job.get( " mapred.mapper.regex " ));
              }
              public   void  map(WritableComparable key, Writable value, OutputCollector output, Reporter reporter)
                     throws  IOException {
                         String text  =  ((Text) value).toString();
                         Matcher matcher  =  pattern.matcher(text);
                       if  (matcher.find()) {
                        output.collect(key, value);
             }
    }
 }

  private  HadoopGrep () {
 }  //  singleton 

public   static   void  main(String[] args)  throws  Exception {
  JobConf grepJob  =   new  JobConf(HadoopGrep. class );
  grepJob.setJobName( " grep-search " );
  grepJob.set( " mapred.mapper.regex " , args[ 2 ]);

  grepJob.setInputPath( new  Path(args[ 0 ]));
  grepJob.setOutputPath( new  Path(args[ 1 ]));
  grepJob.setMapperClass(RegMapper. class );
  grepJob.setReducerClass(IdentityReducer. class );
  JobClient.runJob(grepJob);
 }
}

          RegMapper類的configure()函數接受由main函數傳入的查找字符串，map() 函數進行正則匹配，key是行數，value是文件行的內容，符合的文件行放入中間結果。
        main()函數定義由命令行參數傳入的輸入輸出目錄和匹配字符串，Mapper函數為RegMapper類，Reduce函數是什么都不做，直接把中間結果輸出到最終結果的的IdentityReducer類，運行Job。

整個代碼非常簡單，絲毫沒有分布式編程的任何細節。

三.運行Hadoop程序

        Hadoop這方面的文檔寫得不全面，綜合參考GettingStartedWithHadoop 與Nutch Hadoop Tutorial 兩篇后，再碰了很多釘子才終于完整的跑起來了，記錄如下：     
3.1 local運行模式
       完全不進行任何分布式計算，不動用任何namenode,datanode的做法，適合一開始做調試代碼。
       解壓hadoop，其中conf目錄是配置目錄，hadoop的配置文件在hadoop-default.xml，如果要修改配置，不是直接修改該文件，而是修改hadoop-site.xml，將該屬性在hadoop-site.xml里重新賦值。
       hadoop-default.xml的默認配置已經是local運行，不用任何修改，配置目錄里唯一必須修改的是hadoop-env.sh 里JAVA_HOME的位置。
       將編譯好的HadoopGrep與RegMapper.class 放入hadoop/build/classes/demo/hadoop/目錄 

        或者編譯成jar包HadoopGrep.jar放入hadoop/build/classes/demo/hadoop/目錄

        找一個比較大的xx.log文件放，然后運行

        查看輸出目錄的結果，查看hadoop/logs/里的運行日志。  
        在重新運行前，先刪掉輸出目錄。
 

  3.2 集群運行模式

    （查看集群配置：http://blog.csdn.net/hguisu/article/details/7237395）

      1 ）執行bin/hadoop dfs 可以看到它所支持的文件操作指令。  

      2） 創建目錄輸入inpu：  
           $ bin/hadoop dfs -mkdir input   

      3）上傳文件xx.log到指定目錄 input ：  
           $ bin/hadoop dfs -put xx.log input

       4 )  執行 bin/hadoop demo.hadoop.HadoopGrep input  output
             (jar包運行：bin/hadoop jar HadoopGrep.jar  HadoopGrep  input   /tmp/output  "[a-b]" )

       5 ） 查看輸出文件：

           將輸出文件從分布式文件系統拷貝到本地文件系統查看：
            $ bin/hadoop fs -get output output
            $ cat output/*

            或者
            在分布式文件系統上查看輸出文件：
            $ bin/hadoop fs -cat output/*

            重新執行前,運行hadoop/bin/hadoop dfs -rm output刪除output目錄

       7.運行hadoop/bin/stop-all.sh 結束。
    

四、效率

    經測試，Hadoop并不是萬用靈丹，很取決于文件的大小和數量，處理的復雜度以及群集機器的數量，相連的帶寬，當以上四者并不大時，hadoop優勢并不明顯。
    比如，不用hadoop用java寫的簡單grep函數處理100M的log文件只要4秒，用了hadoop local的方式運行是14秒，用了hadoop單機集群的方式是30秒，用雙機集群10M網口的話更慢，慢到不好意思說出來的地步。