- mapreduce編程實例

　　“平均成績”主要目的還是在重溫經(jīng)典“WordCount”例子，可以說是在基礎(chǔ)上的微變化版，該實例主要就是實現(xiàn)一個計算學生平均成績的例子。

　　3.1 實例描述

　　對輸入文件中數(shù)據(jù)進行就算學生平均成績。輸入文件中的每行內(nèi)容均為一個學生的姓名和他相應的成績，如果有多門學科，則每門學科為一個文件。要求在輸出中每行有兩個間隔的數(shù)據(jù)，其中，第一個代表學生的姓名，第二個代表其平均成績。

　　樣本輸入：

　　1）math：

　　張三 88

　　李四 99

　　王五 66

　　趙六 77

　　2）china：

　　張三 78

　　李四 89

　　王五 96

　　趙六 67

　　3）english：

　　張三 80

　　李四 82

　　王五 84

　　趙六 86

　　樣本輸出：

　　張三 82

　　李四 90

　　王五 82

　　趙六 76

　　3.2 設(shè)計思路

　　計算學生平均成績是一個仿“WordCount”例子，用來重溫一下開發(fā)MapReduce程序的流程。程序包括兩部分的內(nèi)容：Map部分和Reduce部分，分別實現(xiàn)了map和reduce的功能。

　　Map處理的是一個純文本文件， 文件中存放的數(shù)據(jù)時每一行表示一個學生的姓名和他相應一科成績。Mapper處理的數(shù)據(jù)是由InputFormat分解過的數(shù)據(jù)集，其中 InputFormat的作用是將數(shù)據(jù)集切割成小數(shù)據(jù)集InputSplit，每一個InputSlit將由一個Mapper負責處理。此 外，InputFormat中還提供了一個RecordReader的實現(xiàn)，并將一個InputSplit解析成《key，value》對提 供給了map函數(shù)。InputFormat的默認值是TextInputFormat，它針對文本文件，按行將文本切割成InputSlit，并用 LineRecordReader將InputSplit解析成《key，value》對，key是行在文本中的位置，value是文件中的 一行。

　　Map的結(jié)果會通過partion分發(fā)到Reducer，Reducer做完Reduce操作后，將通過以格式OutputFormat輸出。

　　Mapper最終處理的結(jié)果對《key，value》，會送到Reducer中進行合并，合并的時候，有相同key的鍵/值對則送到同一個 Reducer上。Reducer是所有用戶定制Reducer類地基礎(chǔ)，它的輸入是key和這個key對應的所有value的一個迭代器，同時還有 Reducer的上下文。Reduce的結(jié)果由Reducer.Context的write方法輸出到文件中。

　　3.3 程序代碼

　　程序代碼如下所示：

　　package com.hebut.mr;

　　import java.io.IOException;

　　import java.util.Iterator;

　　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.LongWritable;

　　import org.apache.hadoop.io.Text;

　　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.FileInputFormat;

　　import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

　　import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

　　import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

　　import org.apache.hadoop.util.GenericOptionsParser;

　　public class Score {

　　public static class Map extends

　　Mapper《LongWritable， Text， Text， IntWritable》 {

　　// 實現(xiàn)map函數(shù)

　　public void map（LongWritable key， Text value， Context context）

　　throws IOException， InterruptedException {

　　// 將輸入的純文本文件的數(shù)據(jù)轉(zhuǎn)化成String

　　String line = value.toString（）;

　　// 將輸入的數(shù)據(jù)首先按行進行分割

　　StringTokenizer tokenizerArticle = new StringTokenizer（line， “\n”）;

　　// 分別對每一行進行處理

　　while （tokenizerArticle.hasMoreElements（）） {

　　// 每行按空格劃分

　　StringTokenizer tokenizerLine = newStringTokenizer（tokenizerArticle.nextToken（））;

　　String strName = tokenizerLine.nextToken（）;// 學生姓名部分

　　String strScore = tokenizerLine.nextToken（）;// 成績部分

　　Text name = new Text（strName）;

　　int scoreInt = Integer.parseInt（strScore）;

　　// 輸出姓名和成績

　　context.write（name， new IntWritable（scoreInt））;

　　}

　　}

　　}

　　public static class Reduce extends

　　Reducer《Text， IntWritable， Text， IntWritable》 {

　　// 實現(xiàn)reduce函數(shù)

　　public void reduce（Text key， Iterable《IntWritable》 values，

　　Context context） throws IOException， InterruptedException {

　　int sum = 0;

　　int count = 0;

　　Iterator《IntWritable》 iterator = values.iterator（）;

　　while （iterator.hasNext（）） {

　　sum += iterator.next（）.get（）;// 計算總分

　　count++;// 統(tǒng)計總的科目數(shù)

　　}

　　int average = （int） sum / count;// 計算平均成績

　　context.write（key， new IntWritable（average））;

　　}

　　}

　　public static void main（String［］ args） throws Exception {

　　Configuration conf = new Configuration（）;

　　// 這句話很關(guān)鍵

　　conf.set（“mapred.job.tracker”， “192.168.1.2:9001”）;

　　String［］ ioArgs = new String［］ { “score_in”， “score_out” };

　　String［］ otherArgs = new GenericOptionsParser（conf， ioArgs）.getRemainingArgs（）;

　　if （otherArgs.length ！= 2） {

　　System.err.println（“Usage： Score Average 《in》 《out》”）;

　　System.exit（2）;

　　}

　　Job job = new Job（conf， “Score Average”）;

　　job.setJarByClass（Score.class）;

　　// 設(shè)置Map、Combine和Reduce處理類

　　job.setMapperClass（Map.class）;

　　job.setCombinerClass（Reduce.class）;

　　job.setReducerClass（Reduce.class）;

　　// 設(shè)置輸出類型

　　job.setOutputKeyClass（Text.class）;

　　job.setOutputValueClass（IntWritable.class）;

　　// 將輸入的數(shù)據(jù)集分割成小數(shù)據(jù)塊splites，提供一個RecordReder的實現(xiàn)

　　job.setInputFormatClass（TextInputFormat.class）;

　　// 提供一個RecordWriter的實現(xiàn)，負責數(shù)據(jù)輸出

　　job.setOutputFormatClass（TextOutputFormat.class）;

　　// 設(shè)置輸入和輸出目錄

　　FileInputFormat.addInputPath（job， new Path（otherArgs［0］））;

　　FileOutputFormat.setOutputPath（job， new Path（otherArgs［1］））;

　　System.exit（job.waitForCompletion（true） ？ 0 ： 1）;

　　}

　　}

　　3.4 代碼結(jié)果

　　1）準備測試數(shù)據(jù)

　　通過Eclipse下面的“DFS Locations”在“/user/hadoop”目錄下創(chuàng)建輸入文件“score_in”文件夾（備注：“score_out”不需要創(chuàng)建。）如圖3.4-1所示，已經(jīng)成功創(chuàng)建。

　?

　　然后在本地建立三個txt文件，通過Eclipse上傳到“/user/hadoop/score_in”文件夾中，三個txt文件的內(nèi)容如“實例描述”那三個文件一樣。如圖3.4-2所示，成功上傳之后。

　　備注：文本文件的編碼為“UTF-8”，默認為“ANSI”，可以另存為時選擇，不然中文會出現(xiàn)亂碼。

　　從SecureCRT遠處查看“Master.Hadoop”的也能證實我們上傳的三個文件。

　?

　　查看三個文件的內(nèi)容如圖3.4-3所示：

　　?

　　2）查看運行結(jié)果

　　這時我們右擊Eclipse 的“DFS Locations”中“/user/hadoop”文件夾進行刷新，這時會發(fā)現(xiàn)多出一個“score_out”文件夾，且里面有3個文件，然后打開雙 其“part-r-00000”文件，會在Eclipse中間把內(nèi)容顯示出來。如圖3.4-4所示。

　　?