1、MapReduce編程實例：單詞計數本節介紹如何編寫基本的MapReduce程序實現數據分析。本節代碼是基于Hadoop2.7.3開發的。任務準備單詞計數(WordCou nt)的任務是對一組輸入文檔中的單詞進行分別計數。假設文件的量 比較大，每個文檔又包含大量的單詞，則無法使用傳統的線性程序進行處理，而這類問題正 是MapReduce可以發揮優勢的地方。在前面MapReduce實例分析：單詞計數教程中已經介紹了用MapReduce實現單詞 計數的基本思路和具體執行過程。下面將介紹如何編寫具體實現代碼及如何運行程序。首先，在本地創建3個文件：file00l、file002和file003，文件

2、具體內容如表1所示。表1單詞計數輸入文件文件名file001fileO02file003文件內容Hello worldConn ected worldOne worldOne dreamHello HadoopHello MapHello Reduce再使用HDFS命令創建一個in put文件目錄。hadoop fs -mkdir in put然后，把fileOOl、file002和fileOO3上傳到HDFS中的in put目錄下。hadoop fs -put file001 in puthadoop fs -put file002 in puthadoop fs -put file003

3、in put編寫MapReduce程序的第一個任務就是編寫Map程序。在單詞計數任務中，Map需 要完成的任務就是把輸入的文本數據按單詞進行拆分，然后以特定的鍵值對的形式進行輸 出。編寫Map 程序Hadoop MapReduce框架已經在類Mapper中實現了 Map任務的基本功能。為了實 現Map任務，開發者只需要繼承類Mapper，并實現該類的Map函數。為實現單詞計數的Map任務，首先為類Mapper設定好輸入類型和輸出類型。這里， Map函數的輸入是key,value形式，其中，key是輸入文件中一行的行號，value是 該行號對應的一行內容。所以，Map函數的輸入類型為LongWr

4、itable,Text。Map函數的功能為完成文本分割 工作，Map函數的輸出也是key,value形式，其中，key是單詞,value為該單詞出 現的次數。所以，Map函數的輸出類型為Text，LongWritable。以下是單詞計數程序的Map任務的實現代碼。public static class CoreMapper extends MapperObject,Text,Text,IntWritable private static final IntWritable one = new IntWritable(l);private static Text label = new Text

5、();public void map (Object key, Text value,MapperObject,Text,Text,IntWritable Context context)throws IOException,InterruptedException StringTokenizer tokenizer = new StringTokenizer(value.toString();while (tokenizer. hasMoreTokens() label. set(tokenizer. nextToken();context.writ e (label, one);在上述代碼

6、中，實現Map任務的類為CoreMapper。該類首先將需要輸出的兩個變量one和label進行初始化。變量one的初始值直接設置為1,表示某個單詞在文本中出現過。 Map函數的前兩個參數是函數的輸入參數，value為Text類型，是指每次讀入文本的一行，key為Object類型，是指輸入的行數據在文本中的行號。StringTokenizer類機器方法將value變量中文本的一行文字進行拆分，拆分后的單詞放 在tokenizer列表中。然后程序通過循環對每一個單詞進行處理，把單詞放在label中， 把one作為單詞計數。在函數的整個執行過程中，one的值一直是1。在該實例中，key沒有被明顯地

7、使用到。con text是Map函數的一種輸出方式，通過使用該變量，可以直接將中間結果存儲在其中。根據上述代碼，Map任務結束后，3個文件的輸出結果如表2所示。表2單詞計數Map任務輸出結果文件名/MapfileOO1/Map1file002/Map2file003/Map3Map任務輸出結果編寫 Reduce 程序編寫MapReduce程序的第二個任務就是編寫 Reduce程序。在單詞計數任務中，Reduce需要完成的任務就是把輸入結果中的數字序列進行求和從而得到每個單詞的出現 次數。在執行完Map函數之后，會進入Shuffle階段，在這個階段中，MapReduce框架會自 動將Map階段的

8、輸出結果進行排序和分區，然后再分發給相應的Reduce任務去處理。經過Map端Shuffle階段后的結果如表3所示。表3單詞計數Map端Shuffle階段輸出結果文件名/MapfileOO1/Map1file002/Map2fileOO3/Map3Map端Shuffle階段輸出結果world, 0n e, Hello,Reduce端接收到各個Map端發來的數據后，會進行合并，即把同一個key,也就是同一 單詞的鍵值對進行合并，形成key, 形式的輸出。經過Map端Shuffle 階段后的結果如表4所示。表4單詞計數Reduce端Shuffle階段輸出結果Reduce 端Hello,Shuffl

9、e階段輸出結果O ne,world, Reduce階段需要對上述數據進行處理從而得到每個單詞的出現次數。從Reduce函數的 輸入已經可以理解Reduce函數需要完成的工作就是首先對輸入數據value中的數字序 列進行求和。以下是單詞計數程序的Reduce任務的實現代碼。public static class CoreReducer extendsReducerText, IntWritable, Text, IntWritable 2.private IntWritable count 二 new IntWritable ();public void reduce(Text key,Iter

10、ablevalues, Reducer Context context)throwsIOException, InterruptedException intsum = 0;for(IntWritableintWritable : values)sum+=intWritable. get();&count.set (sum);context.writ e (key, cou nt);與Map任務實現相似,Reduce任務也是繼承Hadoop提供的類Reducer并實現其接 口。Reduce函數的輸入、輸出類型與Map函數的輸出類型本質上是相同的。在Reduce函數的開始部分，首先設置sum參數

11、用來記錄每個單詞的出現次數，然后遍 歷value列表，并對其中的數字進行累加，最終就可以得到每個單詞總的出現次數。在輸 出的時候，乃然使用context類型的變量存儲信息。當Reduce階段結束時，就可以得到最終需要的結果，如表5所示。表5單詞計數Reduce任務輸出結果Reduce任務輸出結果編寫main函數為了使用CoreMapper和CoreReducer類進行真正的數據處理，還需要在main函數 中通過Job類設置Hadoop MapReduce程序運行時的環境變量，以下是具體代碼。public static void main(String args) throws Exceptio

12、n Configuration conf = new Configuration();String otherArgs = newGenericOptionsParser(conf, args). getRemainingArgsO;if (otherArgs. length != 2) Sys tem.err.pri ntln( Usage:wordcou ntinout)；Sys tem.ex it( 2);& Job job = new Job (conf, WordCount); /設置環境參數job.se tJarByClass (WordCoun t. class); /設置程序的

13、類名9.10.11.12.13.14.15.16.17.18.19.20.1.2.3.4.5.6.7.&9.10.job.se tM apperClass(CoreMapper. class); /添力口 Mapper 類job. se tReducerClass(CoreReducer. class); /添力口 Reducer 類job. se tOu tpu tKeyClass (Tex t. class); /設置輸出 key 的類型job. setOutputValueClass (IntWritable. class);/設置輸出value的類型FilelnputFormat.ad

14、dInputPath (job, new Path (otherArgs 0);/設置輸入文件路徑FileOutputFormat.setOutputPath (job, new Path (otherArgs 1);/設置輸入文件路徑System.exit(job.waitForCompletion(true) ? 0 : 1);代碼首先檢查參數是不是正確，如果不正確就提醒用戶。隨后，通過Job類設置環境參數， 并設置程序的類、Mapper類和Reducer類。然后，設置了程序的輸出類型，也就是Reduce函數的輸出結果key,value中key和value各自的類型。最后，根據程序運 行時

15、的參數，設置輸入、輸出文件路徑。編寫MapReduce程序需要引用Hadoop的以下幾個核心組件包/它們實現了 HadoopMapReduce 框架。import java. io. IOException;import java. util.StringTokenizer;import org. apache. hadoop. conf. Configuration;import org. apache. hadoop. fs. Path;import org. apache. hadoop. io. IntWritable;import org. apache. hadoop. io. T

16、ext;import org. apache. hadoop. mapreduce. Job;import org. apache. hadoop. mapreduce. Mapper;import org. apache. hadoop. mapreduce. Reducer;import org. apache. hadoop. mapreduce. lib. input. FilelnputFormat;import org.apache.hadoop.mapreduce. lib. output.FileOutputFormat;import org.apache.hadoop. ut

17、il.GenericOptionsParser;這些核心組件包的基本功能描述如表6所示。表6 Hadoop Map Reduce核心組件包的基本功能包功能org.apache.hadoop.c onf定義了系統參數的配置文件處理方法org.apache.hadoop.fs定義了抽象的文件系統APIorg.apache.hadoop.mapreduceHadoop MapReduce框架的實現，包括任務的分發調度等org.apache.hadoop.io定義了通用的I/O API，用于網絡、數據庫和文件數據對象進行讀寫操作運行代碼在運行代碼前需要先把當前工作目錄設置為/user/local/Ha

18、doop。編譯WordCount程 序需要以下3個Jar，為了簡便起見，把這3個Jar添加到CLASSPATH中。$exportCLASSPATH二/usr/local/hadoop/share/hadoop/comm on/hadoop-comm on-jar:$CLASSPATH$exportCLASSPATH二/usr/local/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-jar:$CLAS SPATH$exportCLASSPATH二/usr/local/hadoop/share/hadoop/comm on/lib/comm on-cli-1.2.jar:$CLASSPATH使用JDK包中的工具對代碼進行編譯。$ javac WordCo