本文發表于本人 博客 。

????前面幾次講了關于Hadoop的環境搭建、HDFS操作，今天接著繼續。本來Hadoop源碼中就有一個例子WordCount,但是今天我們來自己實現一個加深對這個Mapper、Reducer的理解，如有不對歡迎指正。

????我們先來梳理一下思路，對于自定義Mapper以及Reducer，我們先要覆蓋其map以及reduce函數，然后按照相關步驟比如設置輸入文件目錄、輸入文件格式化類、設置自定義Mapper、分區、排序、分組、規約、設置自定義Reducer等等。這里我們把輸入文件的使用空格分割(也可以用制表符來)，下面是自定義Mapper類MyMapper：

    import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Mapper.Context;



public class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

    

    @Override

    protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException {

        String[] splied = value.toString().split(" ");

        for (int i = 0; i < splied.length; i++) {

            String lineWord = splied[i];

            context.write(new Text(lineWord), new LongWritable(1));

        }

    }

}
  

這里我選擇的是新的API，相關庫基本是在org.apache.hadoop.mapreduce下，舊API是在org.apache.hadoop.mapred下，包括一些引用庫也是這樣。自定義MyMapper是泛型繼承Mapper，其中參數 key\value 是Hadoop內部類型，它不支持java的基本類型這里我們需要注意下為什么不選擇java的基本類型呢，原因是不需要其它額外是操作，而且本身需要序列化反序列化并提升其性能所以加入了hadoop的類型放棄java的基本類型。關于hadoop key\value 跟java基本類型相互轉換的問題也很簡單，從java基本類型轉換至hadoop的 key\value 的話直接new帶參就可以了，從hadoop的key\value類型轉換至java的基本類型使用get方法就可以了！如：

    LongWritable lw = new LongWritable(1L);

long temp = lw.get();
  

接下來繼續看自定義Reducer類MyReduce：

    import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.Reducer.Context;



public class MyReduce extends Reducer<Text, LongWritable, Text, LongWritable> {



    @Override

    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

        long count = 0L;

        for(LongWritable value: values) {

            count += value.get();

        }

        context.write(key, new LongWritable(count));

    }

}
  

這個跟上面類似了，再來看看main方法的如何執行的！

        

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.InputFormat;

import org.apache.hadoop.mapreduce.Job;

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.mapreduce.lib.partition.HashPartitioner;



import com.sun.org.apache.xpath.internal.axes.HasPositionalPredChecker;



public class Test {

    static final String OUTPUT_DIR = "hdfs://hadoop-master:9000/mapreduce/output/";

    static final String INPUT_DIR = "hdfs://hadoop-master:9000/mapreduce/input/test.txt";

    

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        Job job = new Job(conf, Test.class.getSimpleName());        

        deleteOutputFile(OUTPUT_DIR);

        

        //1設置輸入目錄

        FileInputFormat.setInputPaths(job, INPUT_DIR);

        //2設置輸入格式化類

        job.setInputFormatClass(TextInputFormat.class);

        //3設置自定義Mapper以及鍵值類型

        job.setMapperClass(MyMapper.class);

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(LongWritable.class);

        //4分區

        job.setPartitionerClass(HashPartitioner.class);

        job.setNumReduceTasks(1);

        //5排序分組

        //6設置在自定義Reduce以及鍵值類型

        job.setReducerClass(MyReduce.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(LongWritable.class);

        //7設置輸出目錄

        FileOutputFormat.setOutputPath(job, new Path(OUTPUT_DIR));

        //8提交job

        job.waitForCompletion(true);

    }

    

    static void deleteOutputFile(String path) throws Exception{

        Configuration conf = new Configuration();

        FileSystem fs = FileSystem.get(new URI(INPUT_DIR),conf);

        if(fs.exists(new Path(path))){

            fs.delete(new Path(path));

        }

    }

}
  

執行的時候先會輸出上次執行過的輸出目錄。然后就按照步驟：

    1.設置輸入文件目錄；

2.輸入文件格式化類；

3.設置自定義Mapper以及其鍵值類型；

4.分區；

5.排序；

6.分組；

7.規約；

8.設置自定義Reducer以及其鍵值類型；

9.設置輸出目錄；

10.代碼提交至JobTracker。
  

當然這過程中有些是可以省略的比如輸出文件格式化類。從這個例子我們可以得出：既然可以設置自定義Mapper以及自定義Reducer，那么也應該可以設置自定義的輸入文件格式化類以及分區、排序、分組、規約等等，這個以后會有相關的筆記現在這里只是寫個簡單的例子。我們編寫一個文件如下并把它上傳至hdfs://hadoop-master:9000/mapreduce/input/test.txt：

    luoliang me

asura asura.com luoliang

me
  

然后執行main函數,將會在hdfs://hadoop-master:9000/mapreduce/output/目錄下輸出一個類似part-*的文件，我們可以使用如下命令查看：

    hadoop fs -text /output/part-*
  

此時會輸出：

    asura 1

asura.com 1

luoliang 2

me 2
  

現在文件是輸出了也對比下是正確，但是腦子還是一片空白，不知道其怎么做到的，那么這個就是關于mapreduce的原理了，下面我也說說大概其原理：從把代碼提交至JobTracker開始，它就會從指定的輸入文件路徑去獲取文件，這里支持多個文件以及二級目錄下的多個文件，這里獲取就是使用的HDFS api來操作了！把所有文件讀取出來之后按照指定的大小進行分割InputSplit，把分割好后的鍵值FileSplit（比如：<0,"luoliang me">,<13,"asura asura.com luoliang">）再轉化為RecordReader(比如<"luoliang",1>,<"luoliang",1>)，此時全部轉換完畢后會每個都調用map函數，map函數把數據寫入到Mapper.Context中，再會對數據進行分區排序分組規約，最后通過shuffle到達reduce端，這其中每個map的輸出數量是等于reduce的輸入數量。到達reduce端數據已經發生了質變了不在是<"luoliang",1>而是類似變成<"luoliang",{1,1}>這樣的鍵值數據，這是我們需要迭代獲取總數量并在寫會context中，計算完后輸出到指定的目錄。在這里由于有重復的單詞所以map函數的調用次數跟reduce函數調用次數是不同的。規約這個其實就是自定義reduce，但是這個不是必須有的因為如果是統計關于類似平均數的問題，數據在map端進行規約了，雖然傳送時間以及處理時間減少性能提升了但是對于最終結果可能會有影響，所以這個規約要看具體情況才能使用。至于這個 shuffle 一步還不是怎么了解需要多多再看看。

這次先到這里。堅持記錄點點滴滴！

Hadoop 編寫WordCount