第4周mapr程序設計涉及多語言編程

1、MapReduce程序設計（涉及多語言）講師：董西成OpenPassionValue目錄1. MapReduce的編程模型2. MapReduce編程接口介紹3. Java編程4. 多語言編程5. 總結2MapReduce編程模型ØMapReduce將整個運行過程分為兩個階段：Map階段和Reduce階段Map階段由一定數量的Map Task組成Øüüü：InputFormat輸入數據格式輸入數據處理：Mapper數據分組：PartitionerØReduce階段由一定數量的Reduce Task組成üüü

2、;數據拷貝數據按照key排序數據處理：Reducer數據輸出格式：OutputFormatü3MapReduce編程模型外部物理結構4MapReduce編程模型Ø Map階段ü InputFormat（默認TextInputFormat）ü Mapperü Combiner（local reducer）ü PartitionerØ Reduce階段ü Reducerü OutputFormat（默認TextOutputFormat）5目錄1. MapReduce的編程模型2. MapReduce編程接口介

3、紹3. Java編程4. 多語言編程5. 總結6MapReduce編程接口Ø Hadoop提供了三種編程方式；ü Java（最原始的方式）ü Hadoop Streaming（支持多語言）ü Hadoop Pipes（支持C/C+）Ø Java編程接口是所有編程方式的基礎；Ø 不同的編程接口只是給用戶的形式不同而已，內部執(zhí)行引擎是一樣的；Ø 不同編程方式效率不同。Java編程接口Ø Java編程接口組成；ü 舊APIü 新APIØ 新API具有更好的擴展性；Ø 兩種編程接口只

4、是給用戶的形式不同而已，內部執(zhí)行引擎是一樣的；Ø 舊API可以完全兼容Hadoop 2.0，但新API不行。Hadoop StreamingØ 與Linux管道機制一致Ø 通過標準輸入輸出實現進程間通信Ø 標準輸入輸出是任何語言都有的Ø 幾個舉例：ü cat 1.txt | grep “dong” | sortü cat 1.txt | python grep.py | java sort.jarHadoop Streaming/pipes目錄1. MapReduce的編程模型2. MapReduce編程接口介紹3. Jav

5、a編程4. 多語言編程5. 總結11實例1：WordCount問題WordCount問題map階段WordCount問題reduce階段WordCount問題mapper設計與實現WordCount問題reducer設計與實現WordCount問題main函數設計與實現WordCount問題程序運行WordCount問題程序運行Wordcount問題輸入數據格式l使用默認的TextInputFormatü 每個Map Task處理一個split；ü 一個split大小等于一個block；ü 如果最后一行數據被截斷，則部分；后一個block前半ü 轉換成k

6、ey/value對，key是偏移量，value是行內容。Wordcount問題輸入數據格式Wordcount問題數據流InputFormat輸入數據public interface InputFormat<K, V> InputSplit getSplits(JobConf job, int numSplits) throws IOException;RecordReader<K, V> getRecordReader(InputSplit split,JobConf job,Reporter reporter) throws IOException;l 默認為Text

7、InputFormat，文本文件的；l 用戶可通過參數mapred.input.format.class設置InpuFormat實現Mappermap處理邏輯org.apache.hadoop.mapred（舊API）：public interface Mapper<K1, V1, K2, V2> extends JobConfigurable, Closeable void map(K1 key, V1 value, OutputCollector<K2, V2> output, Reporter reporter)throws IOException;l 新API位

8、于org.apache.hadoop.mapreduce.Mapper中；l 新API更加靈活。Partitionermap輸出結構分片org.apache.hadoop.mapred（舊API）：public interface Partitioner<K2, V2> extends JobConfigurable int getPartition(K2 key, V2 value, int numPartitions);org.apache.hadoop.mapreduce（新API）：public abstract class Partitioner<KEY, VALU

9、E> public abstract int getPartition(KEY key, VALUE value, int numPartitions);Reducerreduce處理邏輯org.apache.hadoop.mapred（舊API）：public interface Reducer<K2, V2, K3, V3> extends JobConfigurable, Closeable void reduce(K2 key, Iterator<V2> values,OutputCollector<K3, V3> output, Report

10、er reporter)throws IOException;l 新API位于org.apache.hadoop.mapreduce.Reducer中；l 新API更加靈活。小結實例2：Grep問題一批TB或者PB量級的文檔，需要完成以下功能：lll搜索符合某種規(guī)則（正則表達式）的單詞或者句子；統計相應的單詞或者句子的數目；按照數目對其進行排序，并輸出最終結果。Grep問題解決思路排序shopA-Z.+HDFS中的文檔頻率統計結果排序結果Grep問題解決思路分為兩個作業(yè)：l作業(yè)1：WordCountü 統計符合某種規(guī)則的單詞數目l作業(yè)2：Sortü 按照單詞數目進行全排序&

11、#252; 依賴于前一個作業(yè)的輸出結果Grep問題解決思路Grep問題解決思路Grep問題解決思路Grep問題解決思路Grep程序運行Grep程序運行目錄1. MapReduce的編程模型2. MapReduce編程接口介紹3. Java編程4. 多語言編程5. 總結37多語言程序設計思路l以標準輸入流作為輸入；ü C+：cinü C：scanfl以標準輸出流作為輸出；ü C+：coutü C：printfl可實現Mapper和Reducer，其他組件（InputFormat、Partitioner等需要用Java語言實現）實例1：用C+實現Wordco

12、unt實例1：用C+實現Wordcount實例1：用C+實現Wordcountllucer實例1：用C+實現WordcountlHadoop上運行Wordcount程序Streaming程序運行方式Streaming程序運行方式說明l “-D”參數：一定要放在所有參數前面，可以有多個；l “-file”或者“-files”參數，設置要分發(fā)到各個節(jié)點上的文件，對于mapper和reducer文件，必須要用或者“-files” 指定。l 每次運行程序前，清空輸出目錄bin/hadoop fs -rmr /test/outputStreaming程序運行方式說明bin/hadoop jar cont

13、rib/streaming/hadoop-streaming-0.20.2-cdh3u6.jar -D mapred.reduce.tasks=10 -D =“wordcount-python” -file ./mapper.py -mapper "python mapper.py" -file ./reducer.py -reducer "python reducer.py" -input /test/input -output /test/output實例2：用PHP實現Wordcount實例2：用PHP實現Wordcount實例2：用PHP實現Wordcountl 測試mapper和reducerü cat test.txt| php mapper.php | sort | php reducer.phpl 在Hadoop上運行Java與Streaming編程方式比較lJava編程ü Hado