第4周mapr程序設(shè)計(jì)涉及多語(yǔ)言編程

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

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

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

4、是給用戶(hù)的形式不同而已，內(nèi)部執(zhí)行引擎是一樣的；Ø 舊API可以完全兼容Hadoop 2.0，但新API不行。Hadoop StreamingØ 與Linux管道機(jī)制一致Ø 通過(guò)標(biāo)準(zhǔn)輸入輸出實(shí)現(xiàn)進(jìn)程間通信Ø 標(biāo)準(zhǔn)輸入輸出是任何語(yǔ)言都有的Ø 幾個(gè)舉例：ü 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. 多語(yǔ)言編程5. 總結(jié)11實(shí)例1：WordCount問(wèn)題WordCount問(wèn)題map階段WordCount問(wèn)題reduce階段WordCount問(wèn)題mapper設(shè)計(jì)與實(shí)現(xiàn)WordCount問(wèn)題reducer設(shè)計(jì)與實(shí)現(xiàn)WordCount問(wèn)題main函數(shù)設(shè)計(jì)與實(shí)現(xiàn)WordCount問(wèn)題程序運(yùn)行WordCount問(wèn)題程序運(yùn)行Wordcount問(wèn)題輸入數(shù)據(jù)格式l使用默認(rèn)的TextInputFormatü 每個(gè)Map Task處理一個(gè)split；ü 一個(gè)split大小等于一個(gè)block；ü 如果最后一行數(shù)據(jù)被截?cái)啵瑒t部分；后一個(gè)block前半ü 轉(zhuǎn)換成k

6、ey/value對(duì)，key是偏移量，value是行內(nèi)容。Wordcount問(wèn)題輸入數(shù)據(jù)格式Wordcount問(wèn)題數(shù)據(jù)流InputFormat輸入數(shù)據(jù)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 默認(rèn)為T(mén)ext

7、InputFormat，文本文件的；l 用戶(hù)可通過(guò)參數(shù)mapred.input.format.class設(shè)置InpuFormat實(shí)現(xiàn)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輸出結(jié)構(gòu)分片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更加靈活。小結(jié)實(shí)例2：Grep問(wèn)題一批TB或者PB量級(jí)的文檔，需要完成以下功能：lll搜索符合某種規(guī)則（正則表達(dá)式）的單詞或者句子；統(tǒng)計(jì)相應(yīng)的單詞或者句子的數(shù)目；按照數(shù)目對(duì)其進(jìn)行排序，并輸出最終結(jié)果。Grep問(wèn)題解決思路排序shopA-Z.+HDFS中的文檔頻率統(tǒng)計(jì)結(jié)果排序結(jié)果Grep問(wèn)題解決思路分為兩個(gè)作業(yè)：l作業(yè)1：WordCountü 統(tǒng)計(jì)符合某種規(guī)則的單詞數(shù)目l作業(yè)2：Sortü 按照單詞數(shù)目進(jìn)行全排序&

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

12、unt實(shí)例1：用C+實(shí)現(xiàn)Wordcount實(shí)例1：用C+實(shí)現(xiàn)Wordcountllucer實(shí)例1：用C+實(shí)現(xiàn)WordcountlHadoop上運(yùn)行Wordcount程序Streaming程序運(yùn)行方式Streaming程序運(yùn)行方式說(shuō)明l “-D”參數(shù)：一定要放在所有參數(shù)前面，可以有多個(gè)；l “-file”或者“-files”參數(shù)，設(shè)置要分發(fā)到各個(gè)節(jié)點(diǎn)上的文件，對(duì)于mapper和reducer文件，必須要用或者“-files” 指定。l 每次運(yùn)行程序前，清空輸出目錄bin/hadoop fs -rmr /test/outputStreaming程序運(yùn)行方式說(shuō)明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實(shí)例2：用PHP實(shí)現(xiàn)Wordcount實(shí)例2：用PHP實(shí)現(xiàn)Wordcount實(shí)例2：用PHP實(shí)現(xiàn)Wordcountl 測(cè)試mapper和reducerü cat test.txt| php mapper.php | sort | php reducer.phpl 在Hadoop上運(yùn)行Java與Streaming編程方式比較lJava編程ü Hado