Hadoop JAVA 開發說明

http://blog.sina.com.cn/s/blog_68b4c68f0101447y.html

 

做爲Hadoop程序員，他要作的事情就是：

一、定義Mapper，處理輸入的Key-Value對，輸出中間結果。
二、定義Reducer，可選，對中間結果進行規約，輸出最終結果。
三、定義InputFormat 和OutputFormat，可選，InputFormat將每行輸入文件的內容轉換爲Java類供Mapper函數使用，不定義時默認爲String。
四、定義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 。

2、程序員編寫的代碼

咱們作一個簡單的分佈式的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/目錄 找一個比較大的log文件放入一個目錄，而後運行

hadoop / bin / hadoop demo.hadoop.HadoopGrep log文件所在目錄 任意的輸出目錄 grep的字符串

查看輸出目錄的結果，查看hadoop/logs/裏的運行日誌。 
在從新運行前，先刪掉輸出目錄。

3.2 單機集羣運行模式

如今來搞一下只有單機的集羣.假設以完成3.1中的設置，本機名爲hadoopserver
第1步. 而後修改hadoop-site.xml ，加入以下內容：

< property > 
< name > fs.default.name </ name > 
< value > hadoopserver:9000 </ value > 
</ property > 
< property > 
< name > mapred.job.tracker </ name > 
< value > hadoopserver:9001 </ value > 
</ property > 
< property > 
< name > dfs.replication </ name > 
< value > 1 </ value > 
</ property >

今後就將運行從local文件系統轉向了hadoop的hdfs系統，mapreduce的jobtracker也從local的進程內操做變成了分佈式的任務系統，9000，9001兩個端口號是隨便選擇的兩個空餘端口號。

另外，若是你的/tmp目錄不夠大，可能還要修改hadoop.tmp.dir屬性。

第2步. 增長ssh不輸入密碼便可登錄。

由於Hadoop須要不用輸入密碼的ssh來進行調度，在不su的狀態下，在本身的home目錄運行ssh-keygen -t rsa ,而後一路回車生成密鑰，再進入.ssh目錄,cp id_rsa.pub authorized_keys
詳細能夠man 一下ssh, 此時執行ssh hadoopserver，不須要輸入任何密碼就能進入了。

3.格式化namenode，執行
bin/hadoop namenode -format

4.啓動Hadoop
執行hadoop/bin/start-all.sh, 在本機啓動namenode,datanode,jobtracker,tasktracker

5.如今將待查找的log文件放入hdfs,。
執行hadoop/bin/hadoop dfs 能夠看到它所支持的文件操做指令。
執行hadoop/bin/hadoop dfs put log文件所在目錄 in ，則log文件目錄已放入hdfs的/user/user-name/in 目錄中

6.如今來執行Grep操做
hadoop/bin/hadoop demo.hadoop.HadoopGrep in out
查看hadoop/logs/裏的運行日誌，從新執行前。運行hadoop/bin/hadoop dfs rmr out 刪除out目錄。

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

3.3 集羣運行模式
假設已執行完3.2的配置，假設第2臺機器名是hadoopserver2
1.建立與hadoopserver一樣的執行用戶，將hadoop解壓到相同的目錄。

2.一樣的修改haoop-env.sh中的JAVA_HOME 及修改與3.2一樣的hadoop-site.xml

3. 將hadoopserver中的/home/username/.ssh/authorized_keys 複製到hadoopserver2,保證hadoopserver能夠無需密碼登錄hadoopserver2
scp /home/username/.ssh/authorized_keys username@hadoopserver2:/home/username/.ssh/authorized_keys 

4.修改hadoop-server的hadoop/conf/slaves文件, 增長集羣的節點，將localhost改成
hadoop-server
hadoop-server2

5.在hadoop-server執行hadoop/bin/start-all.sh
將會在hadoop-server啓動namenode,datanode,jobtracker,tasktracker
在hadoop-server2啓動datanode 和tasktracker

6.如今來執行Grep操做
hadoop/bin/hadoop demo.hadoop.HadoopGrep in out
從新執行前,運行hadoop/bin/hadoop dfs rmr out 刪除out目錄

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