Map Reduce Programming

Presentation on theme: "Map Reduce Programming"— Presentation transcript:

1 Map Reduce Programming
王耀聰 陳威宇 國家高速網路與計算中心(NCHC)

2 Outline 概念 程式基本框架及執行步驟方法 範例一: 範例二:
Hadoop 的 Hello World => Word Count 說明 動手做 範例二: 進階版=> Word Count 2

3 概念 MapReduce 圖解

4 概念 MapReduce in Parallel

5 Program Prototype 程式基本 框架 Map 程式碼 Reduce 程式碼 其他的設定參數程式碼 Class MR{
Class Map …{ } Class Reduce …{ main(){ JobConf conf = new JobConf(“MR.class”); conf.setInputPath(“the_path_of_HDFS ”); conf.setOutputPath(“the_path_of_HDFS ”); conf.setMapperClass(Map.class); conf.setReduceClass(Reduce.class); JobClient.runJob(conf); }} Map 區 Map 程式碼 Reduce 區 Reduce 程式碼 設定區 其他的設定參數程式碼

6 Process Prototype 編譯 jar -cvf MyJar.jar -C MyJava . 執行流程 基本步驟
javac -classpath hadoop-*-core.jar -d MyJava MyCode.java 封裝 jar -cvf MyJar.jar -C MyJava . 執行 bin/hadoop jar MyJar.jar MyCode HDFS_Input/ HDFS_Output/ 所在的執行目錄為Hadoop_Home ./MyJava = 編譯後程式碼目錄 Myjar.jar = 封裝後的編譯檔 先放些文件檔到HDFS上的input目錄 ./input; ./ouput = hdfs的輸入、輸出目錄

7 Word Count Sample (1) 範例 程式一
class MapClass extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map( LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { String line = ((Text) value).toString(); StringTokenizer itr = new StringTokenizer(line); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); output.collect(word, one); }}} 1 2 3 4 5 6 7 8 9

8 Word Count Sample (2) 範例 程式一
class ReduceClass extends MapReduceBase implements Reducer< Text, IntWritable, Text, IntWritable> { IntWritable SumValue = new IntWritable(); public void reduce( Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { int sum = 0; while (values.hasNext()) sum += values.next().get(); SumValue.set(sum); output.collect(key, SumValue); }} 1 2 3 4 5 6 7 8

9 Word Count Sample (3) 範例 程式一 conf.setCombinerClass(Reduce.class);
Class WordCount{ main() JobConf conf = new JobConf(WordCount.class); conf.setJobName("wordcount"); // set path conf.setInputPath(new Path(args[0])); conf.setOutputPath(new Path(args[1])); // set map reduce conf.setOutputKeyClass(Text.class); // set every word as key conf.setOutputValueClass(IntWritable.class); // set 1 as value conf.setMapperClass(MapClass.class); conf.setCombinerClass(Reduce.class); conf.setReducerClass(ReduceClass.class); onf.setInputFormat(TextInputFormat.class); conf.setOutputFormat(TextOutputFormat.class); // run JobClient.runJob(conf); }}

10 核心 Mapper <key/value > 的映射集合 設定 每次map的輸入 map完後的輸出
範例一 說明 核心 Mapper <key/value > 的映射集合 設定 conf.setMapperClass(MapClass.class); 每次map的輸入 map ( WritableComparable, Writable, OutputCollector, Reporter) map完後的輸出 OutputCollector.collect ( WritableComparable,Writable )

11 範例一 說明 核心Combiner 指定一個combiner，它負責對中間過程的輸出進行本地的聚集，這會有助於降低從Mapper 到 Reducer數據傳輸量。 設定 JobConf.setCombinerClass(Class)

12 核心Reducer 將Map送來的<key/value > ,對每個key作value的整合
範例一 說明 核心Reducer 將Map送來的<key/value > ,對每個key作value的整合 輸入: <key, (list of values)> Reduce (WritableComparable, Iterator, OutputCollector, Reporter) 輸出 OutputCollector.collect(WritableComparable, Writable) 若沒有Reduce要執行,可以不編寫

13 配置JobConf 範例一 說明 Hadoop程式架構內主要的執行設定類別
指定Mapper、Combiner、 Partitioner、Reducer、InputFormat和OutputFormat 的類別為何 指定輸入文件 setInputPaths(JobConf, Path...) / addInputPath(JobConf, Path)) 指定輸出文件 setOutputPath(Path) debug script setMapDebugScript(String) / setReduceDebugScript(String) 最多的嘗試次數 setMaxMapAttempts(int) / setMaxReduceAttempts(int)) 容許任務失敗的百分比 setMaxMapTaskFailuresPercent(int) / setMaxReduceTaskFailuresPercent(int) ……

14 任務執行 runJob(JobConf )： submitJob(JobConf )：
範例一 說明 任務執行 runJob(JobConf )： 提交作業，僅當作業完成時返回。 submitJob(JobConf )： 只提交作業，之後需要你輪詢它返回的 RunningJob 句柄的狀態，並根據情況調度。 JobConf.setJobEndNotificationURI(String )： 設置一個作業完成通知，可避免輪詢。

15 WordCount練習 (前置) 範例一 動手做 cd $HADOOP_HOME bin/hadoop dfs -mkdir input
echo "I like NCHC Cloud Course." > input1 echo "I like nchc Cloud Course, and we enjoy this crouse." > input2 bin/hadoop dfs -put input1 input bin/hadoop dfs -put input2 input bin/hadoop dfs -ls input 編輯WordCount.java mkdir MyJava

16 WordCount練習 (執行) 編譯 jar -cvf wordcount.jar -C MyJava . 範例一 動手做
javac -classpath hadoop-*-core.jar -d MyJava WordCount.java 封裝 jar -cvf wordcount.jar -C MyJava . 執行 bin/hadoop jar wordcount.jar WordCount input/ output/ 所在的執行目錄為Hadoop_Home ./MyJava = 編譯後程式碼目錄 wordcount.jar = 封裝後的編譯檔 先放些文件檔到HDFS上的input目錄 ./input; ./ouput = hdfs的輸入、輸出目錄

17 範例一 動手做 WordCount練習 (執行)

18 範例一 動手做 WordCount練習 (結果)

19 WordCount 進階版 WordCount2 功能 步驟 (接續 WordCount 的環境) 範例二 動手做 不計標點符號 不管大小寫
功能 不計標點符號 不管大小寫 步驟 (接續 WordCount 的環境) echo "\." >pattern.txt && echo "\," >>pattern.txt bin/hadoop dfs -put pattern.txt ./ mkdir MyJava2 javac -classpath hadoop-*-core.jar -d MyJava2 WordCount2.java jar -cvf wordcount2.jar -C MyJava2 .

20 範例二 動手做 不計標點符號 執行 bin/hadoop jar wordcount2.jar WordCount2 input output2 -skip pattern.txt dfs -cat output2/part-00000

21 範例二 動手做 不管大小寫 執行 bin/hadoop jar wordcount2.jar WordCount2 -Dwordcount.case.sensitive=false input output3 -skip pattern.txt

22 Tool 處理Hadoop命令執行的選項 透過介面交由程式處理 ToolRunner.run(Tool, String[]) 範例二
補充說明 Tool 處理Hadoop命令執行的選項 -conf <configuration file> -D <property=value> -fs <local|namenode:port> -jt <local|jobtracker:port> 透過介面交由程式處理 ToolRunner.run(Tool, String[])

23 DistributedCache 設定特定有應用到相關的、超大檔案、或只用來參考卻不加入到分析目錄的檔案
範例二 補充說明 DistributedCache 設定特定有應用到相關的、超大檔案、或只用來參考卻不加入到分析目錄的檔案 如pattern.txt檔 DistributedCache.addCacheFile(URI,conf) URI＝hdfs://host:port/FilePath