MapReduce編程實例：單詞計數(shù)

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

2、具體內(nèi)容如表1所示。表1單詞計數(shù)輸入文件文件名file001fileO02file003文件內(nèi)容Hello worldConn ected worldOne worldOne dreamHello HadoopHello MapHello Reduce再使用HDFS命令創(chuàng)建一個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程序的第一個任務(wù)就是編寫Map程序。在單詞計數(shù)任務(wù)中，Map需 要完成的任務(wù)就是把輸入的文本數(shù)據(jù)按單詞進行拆分，然后以特定的鍵值對的形式進行輸 出。編寫Map 程序Hadoop MapReduce框架已經(jīng)在類Mapper中實現(xiàn)了 Map任務(wù)的基本功能。為了實 現(xiàn)Map任務(wù)，開發(fā)者只需要繼承類Mapper，并實現(xiàn)該類的Map函數(shù)。為實現(xiàn)單詞計數(shù)的Map任務(wù)，首先為類Mapper設(shè)定好輸入類型和輸出類型。這里， Map函數(shù)的輸入是key,value形式，其中，key是輸入文件中一行的行號，value是 該行號對應(yīng)的一行內(nèi)容。所以，Map函數(shù)的輸入類型為LongWr

4、itable,Text。Map函數(shù)的功能為完成文本分割 工作，Map函數(shù)的輸出也是key,value形式，其中，key是單詞,value為該單詞出 現(xiàn)的次數(shù)。所以，Map函數(shù)的輸出類型為Text，LongWritable。以下是單詞計數(shù)程序的Map任務(wù)的實現(xiàn)代碼。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、中，實現(xiàn)Map任務(wù)的類為CoreMapper。該類首先將需要輸出的兩個變量one和label進行初始化。變量one的初始值直接設(shè)置為1,表示某個單詞在文本中出現(xiàn)過。 Map函數(shù)的前兩個參數(shù)是函數(shù)的輸入?yún)?shù)，value為Text類型，是指每次讀入文本的一行，key為Object類型，是指輸入的行數(shù)據(jù)在文本中的行號。StringTokenizer類機器方法將value變量中文本的一行文字進行拆分，拆分后的單詞放 在tokenizer列表中。然后程序通過循環(huán)對每一個單詞進行處理，把單詞放在label中， 把one作為單詞計數(shù)。在函數(shù)的整個執(zhí)行過程中，one的值一直是1。在該實例中，key沒有被明顯地

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

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

9、e階段輸出結(jié)果O ne,world, Reduce階段需要對上述數(shù)據(jù)進行處理從而得到每個單詞的出現(xiàn)次數(shù)。從Reduce函數(shù)的 輸入已經(jīng)可以理解Reduce函數(shù)需要完成的工作就是首先對輸入數(shù)據(jù)value中的數(shù)字序 列進行求和。以下是單詞計數(shù)程序的Reduce任務(wù)的實現(xiàn)代碼。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任務(wù)實現(xiàn)相似,Reduce任務(wù)也是繼承Hadoop提供的類Reducer并實現(xiàn)其接 口。Reduce函數(shù)的輸入、輸出類型與Map函數(shù)的輸出類型本質(zhì)上是相同的。在Reduce函數(shù)的開始部分，首先設(shè)置sum參數(shù)

11、用來記錄每個單詞的出現(xiàn)次數(shù)，然后遍 歷value列表，并對其中的數(shù)字進行累加，最終就可以得到每個單詞總的出現(xiàn)次數(shù)。在輸 出的時候，乃然使用context類型的變量存儲信息。當Reduce階段結(jié)束時，就可以得到最終需要的結(jié)果，如表5所示。表5單詞計數(shù)Reduce任務(wù)輸出結(jié)果Reduce任務(wù)輸出結(jié)果編寫main函數(shù)為了使用CoreMapper和CoreReducer類進行真正的數(shù)據(jù)處理，還需要在main函數(shù) 中通過Job類設(shè)置Hadoop MapReduce程序運行時的環(huán)境變量，以下是具體代碼。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); /設(shè)置環(huán)境參數(shù)job.se tJarByClass (WordCoun t. class); /設(shè)置程序的

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); /設(shè)置輸出 key 的類型job. setOutputValueClass (IntWritable. class);/設(shè)置輸出value的類型FilelnputFormat.ad

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

15、的參數(shù)，設(shè)置輸入、輸出文件路徑。編寫MapReduce程序需要引用Hadoop的以下幾個核心組件包/它們實現(xiàn)了 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定義了系統(tǒng)參數(shù)的配置文件處理方法org.apache.hadoop.fs定義了抽象的文件系統(tǒng)APIorg.apache.hadoop.mapreduceHadoop MapReduce框架的實現(xiàn)，包括任務(wù)的分發(fā)調(diào)度等org.apache.hadoop.io定義了通用的I/O API，用于網(wǎng)絡(luò)、數(shù)據(jù)庫和文件數(shù)據(jù)對象進行讀寫操作運行代碼在運行代碼前需要先把當前工作目錄設(shè)置為/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