實驗環境

Hadoop版本：Hadoop2.7.3
linux版本：Ubuntu
JDK版本：JDK1.7

實驗步驟

1. 設定HADOOP的PATH和HADOOP CLASSPATH(這裡假設java的相關路徑已經配置好)

export HADOOP_HOME=/home/luchi/Hadoop/hadoop-2.7.3
export PATH=${HADOOP_HOME}/bin:$PATH
export HADOOP_CLASSPATH=${JAVA_HOME}/lib/tools.jar

1. 編輯CountWord.java檔案

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;

public class WordCount {

  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    private final static IntWritable one = new IntWritable(1
 
);
    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());
        context.write(word, one);
      }
    }
  }

  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(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);
  }
}

1. 編譯WordCount.java
   注意這裡使用的是Hadoop下面的javac進行編譯的：
   首先轉到WordCount.java所在的目錄（如果沒有轉到該子目錄，使用路徑名的話，那麼也就相當於編譯一個java包，這裡需要在上面的程式碼上加上相應的包名）

$ hadoop com.sun.tools.javac.Main WordCount.java

使用這個命令之後，會生成三個檔案：
WordCountIntSumReducer.classWordCountTokenizerMapper.class
WordCount.class
第一個和第二個其實是WordCount類的兩個子靜態類，中間用$表示隸屬關係
4. 將.class檔案打包成jar包

jar cf wc.jar WordCount*.class

注意，這裡用的是JDK1.7的jar命令，區別於後文的hadoop jar命令
5. 在HDFS檔案系統上建立input目錄，作為輸入檔案目錄
我是在/user目錄下建立了一個input子目錄，下面有兩個檔案file1和file2：

luchi@ubuntu:~/Desktop$ hdfs dfs -ls /user/input
Found 2 items
-rw-r--r--   1 luchi supergroup         68 2016-10-16 09:40 /user/input/file1
-rw-r--r--   1 luchi supergroup         15 2016-10-16 09:40 /user/input/file2

file1和file2的檔案是：

luchi@ubuntu:~/Desktop$ hdfs dfs -cat  /user/input/file1 /user/input/file2
i have a dream 
that everyone should chase his dream 
and be honest
 this is cat 2

注意：這裡不需要建立output目錄，因為在執行完map-reduce工作之後，會自動生成該目錄
6. 執行map-reduce工作

hadoop jar wc.jar WordCount /user/input /user/output

需要注意的是，如果WordCount有包名，那麼需要在上述命令裡面的WordCount前面加上包名
執行成功的顯示如下：

16/10/16 10:45:53 INFO mapreduce.Job: Job job_local98859034_0001 completed successfully
16/10/16 10:45:53 INFO mapreduce.Job: Counters: 35
    File System Counters
        FILE: Number of bytes read=10886
        FILE: Number of bytes written=850552
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=234
        HDFS: Number of bytes written=106
        HDFS: Number of read operations=22
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=5
    Map-Reduce Framework
        Map input records=4
        Map output records=17
        Map output bytes=148
        Map output materialized bytes=182
        Input split bytes=206
        Combine input records=17
        Combine output records=16
        Reduce input groups=16
        Reduce shuffle bytes=182
        Reduce input records=16
        Reduce output records=16
        Spilled Records=32
        Shuffled Maps =2
        Failed Shuffles=0
        Merged Map outputs=2
        GC time elapsed (ms)=105
        Total committed heap usage (bytes)=455553024
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=83
    File Output Format Counters 
        Bytes Written=106

從上面的程式碼可以看出，了4個map輸入，產生了17個map輸出，17個combine輸入產生了16個combine輸出，16個reduce輸入產生了16個reduce輸出。注意這裡的map輸入和map輸出以及reduce輸入和reduce輸出指的並不是map和reduce的個數，實際上，FileInputFormat是一個基類，如果不加指定的話，預設的是TextInputFormat，是根據input目錄下的檔案數來確定map的個數的（對於檔案小於block size的情況，對於檔案大小大於block的情況則要再分割槽），這裡的input資料夾下面有兩個檔案，因此，這個任務的job的map個數為2，另一方面，這個hadoop環境是偽分散式的，因此其reduce的個數是1。

1. 看一眼我們的輸出結果檔案：

luchi@ubuntu:~/Desktop$ hdfs dfs -cat /user/output/part-r-00000
2   1
a   1
and 1
be  1
cat 1
chase   1
dream   2
everyone    1
have    1
his 1
honest  1
i   1
is  1
should  1
that    1
this    1

可以看出，詞頻已經統計完畢了。

程式碼和結果分析

說明：這是一個HADOOP新手的理解，可能有些地方不對，不對的地方還請指正。
詳細的API呼叫不準備說了，看官方文件即可，因為剛開始學HADOOP，所以還是想把程式碼執行的經過說下。

程式碼中TokenizerMapper繼承了Mapper這個父類，Mapper這個類是用來進行map操作的，以輸入文字為例，程式碼：

    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

這兩句指定了輸入與輸出的路徑，本例子中指定的路徑是、user/input，接受到這個輸入之後，Map-Reduce框架會首先將文字分塊，然後每塊對應一個map，這裡用了4個map輸入（指的是有四行文字），每個map是以鍵值對對應的，key為文字索引位置，value為一行文字的文字，以下面為例(僅僅是示例，沒有實際意義)：

（0，"i have"）
(24,"dream")

然後每個map的每個項輸入會送入Mapper類，進行map操作，這裡使用的是TokenizerMapper這個Mapper子類進行操作。在本例中：

public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{

    private final static IntWritable one = new IntWritable(1);
    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());
        context.write(word, one);
      }
    }
  }

TokenizerMapper的作用是，將每個map傳來的輸入（鍵值對）進行處理，我們對傳進來的傳進來的鍵值對的key並不關心，只有value有作用，上面程式碼的map方法前兩個引數就是這個意思，我們需要對value進行分詞處理，這裡使用的StringTokenizer這個類進行處理，分詞之後，按照（單詞，詞頻）的鍵值對儲存在context中，並作為輸出的map，這也是為什麼2個map輸入會用多個輸出map

在這之後，程式碼中使用了combine操作，其實combine操作買就是區域性的reduce,因為在map和reduce之間傳遞需要頻寬，因此combine先是把本地的輸出map進行了區域性的reduce，比如說，一個map的輸出是：
（dream,1）
（dream,1）
經過combine之後的輸出是（dream，2），這就減少了傳輸資料量，減少了頻寬的壓力。

之後就是reduce操作，當然在reduce之前還是有sort和shuffle過程，shuffle就是把相似的mapper對映到一個reduce塊中。本例中reduce是使用IntSumReducer這個類實現的，IntSumReducer繼承了Reducer這個父類，和這點Mapper一樣。

public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

這裡的reduce方法和之前的map方法還是有差別的，reduce接受的輸入是key一致的，而值的大小不一樣的輸入，在本例中，key就是word,而values就是該詞的詞頻列表，我們將其相加就得到了這個詞的詞頻。該程式整體的框架如下圖：

在使用了map方法之後， 本例之中是指TokenizerMapper的map方法，得到17個map，本例子只有兩個相同的詞就是dream,經過combine之後，將兩個（dream，1）合併為（combine，2），得到16個map實際輸出,然後這16個輸出map經過shuffle和sort送入reduce過程，此時沒有單詞需要reduce，因為在combine過程中已經將完全一樣的詞dream合併過了，所以16個reduce輸入得到16個reduce輸出，然後就得到了結果。

總結

程式碼執行起來也容易，但是還是有些細節需要探究的，根據這個實驗，我粗略的瞭解了map-reduce的整個過程，當然可能存在一些不當之處，還是需要今後多多學習來補充，歡迎指正。

參考資料：《Hadoop權威指南》