例項描述

　　計算出檔案中每個單詞的頻數，要求輸出結果按照單詞的字母順序進行排序，按照key-value格式輸出結果。

比如輸入檔案為：
    hello world
    hello hadoop
    hello mapreduce

輸出檔案為：
    hadoop 1
    hello 3
    mapreduce 1
    world 1

設計思路

   就是將檔案內容切分成單詞，然後將所有相同的單詞聚集在一起，最後計算單詞出現次數並輸出。
根據MapReduce並行程式設計原則可知，內容切分步驟和資料不相關，可以並行化處理，每個獲得原始
資料的機器只要將資料切分成單詞就行。所以可以在Map階段完成單詞切分任務。另外，相同的單詞
頻數計算也可以進行並行處理。例項來看，不同單詞之間的頻數不相關。所以可以將相同的單詞交給
同一臺機器進行處理，然後輸出結果。這個過程可以在Reduce過程完成。將中間結果在根據不同的
單詞分組分發給不同的機器。
 

• Map階段：完成由輸入資料到單詞切分工作
• shuffle階段：完成相同單詞的聚集和分發工作
• Reduce階段：負責接收所有單詞，並計數

MapReduce中資料傳遞都是key-value形式。shuffle排序聚集分發是按照key進行排序的

執行步驟

使用eclipse作為編譯工具的：

    1，新建MapReduce專案
    2，檢查MapReduce jar包集全不
    3，寫程式碼
    4，配置執行引數

選擇run Configurations

檢查Mian方法路徑

選擇引數欄，第一個為輸入檔案的位置(Liunx上執行支援萬用字元匹配路徑)，第二個為輸出檔案位置，在run就能執行了，執行完畢，結果在對應的輸出引數路徑。

使用命令列執行的

1.先新建一個java檔案
2.寫好程式
3.編譯程式
    對於hadoop-1.x的編譯指令為(僅為樣例，版本不同，要改版本號)

javac -classpath hadoop-1.0.1-core.jar:lib/commons-cli-1.2.jar -d WordCount WordCount.java

對於hadoop-2.x的編譯指令為(僅為樣例，版本不同，要改版本號)

javac -classpath $HADOOP_HOME/share/hadoop/common/hadoop-common-2.6.0.jar:$
HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.0.jar
 -d /opt/hadoop-2.6.0-cdh5.6.0/MapReduceClass  WordCount.java
 

這是因為hadoop-2.x版本沒有core jar了，被拆分成多個jar了

打包成jar

jar -cvf wordcount.jar -C WordCount

叢集上執行程式

bin/hadoop jar wordcount.jar WordCount wordcount_input wordcount_output

檢視結果

bin/hadoop fs -cat wordcount_output/part-r-00000

WordCount程式碼

package WordCount;

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.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.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


public class WordCount {

//繼承Mapper介面，設定Mapper輸入型別為 Object , Text>
//輸出型別為<Text,IntWritable>
public static class TokenizerMapper extends  Mapper<Object,Text,Text,IntWritable>{
        //one 表示單詞出現一次
        private final static IntWritable one = new IntWritable(1);
        //word用於儲存切下的單詞
        private Text word = new Text();
        public void map(Object key,Text value,Context context) throws IOException,InterruptedException{
            StringTokenizer itr = new StringTokenizer(value.toString());
            while(itr.hasMoreTokens()){
                word.set(itr.nextToken()); //切下的單詞存入word
                context.write(word,one);
            }
        }
}

    //繼承Reducer介面，設定Reduce的輸入型別為<Text,IntWritable>
    //輸出型別為<Text,IntWritable>
public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
    //result記錄單詞的頻數
    private IntWritable result = new IntWritable();

    public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException,InterruptedException{
        int sum = 0;

        //對獲取的<key,value-list>計算value的和
        for(IntWritable val : values)
        {
        sum+=val.get();
        }
        //將頻數設定到result
        result.set(sum);
        //收集結果
        context.write(key,result);
    }
}
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
//      conf.set("fs.hdfs.impl",org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
        String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
        if(otherArgs.length != 2){
            System.err.println("Usage:wordcount <in> <out>");
            System.exit(2);
        }
        Job job = new Job(conf,"word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        System.exit(job.waitForCompletion(true)?0:1);


    }
}

總結

   對於wordcount這個程式，開啟了我MapReduce程式設計的第一步吧，最開始，看的書是1.0.1版本的
但是自己的環境是2.6.5版本的，在指令上，一直執行問題。後面查了查知道，hadoop-2.x，已經
沒有core jar了。命令列執行一直失敗。後來在進行改進，所以這個問題是很必要注意的