關於MapReduce單詞統計的例子:

要統計的文件的文件名為hello

hello中的內容如下

hello　　you

hello　　me

通過MapReduce程序統計出文件中的各個單詞出現了幾次.(兩個單詞之間通過tab鍵進行的分割)

import java.io.IOException;

import mapreduce.WordCountApp.WordCountMapper.WordCountReducer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.output.FileOutputFormat;

/**
 * 以文本
 * hello    you
 * hello    me
 * 為例子.
 * map方法調用了兩次,因為有兩行
 * k2 v2 鍵值對的數量有幾個?
 * 有4個.有四個單詞.
 * 
 * 會產生幾個分組?
 * 產生3個分組.
 * 有3個不同的單詞.
 *
 */
public class WordCountApp {
    public static void main(String[] args) throws Exception {
        //程序在這里運行,要有驅動.
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf,WordCountApp.class.getSimpleName());
        
        //我們運行此程序通過運行jar包來執行.一定要有這句話.
        job.setJarByClass(WordCountApp.class);
        
        FileInputFormat.setInputPaths(job,args[0]);
        
        job.setMapperClass(WordCountMapper.class);//設置Map類
        job.setMapOutputKeyClass(Text.class);//設置Map的key
        job.setMapOutputValueClass(LongWritable.class);//設置Map的value
        job.setReducerClass(WordCountReducer.class);//設置Reduce的類
        job.setOutputKeyClass(Text.class);//設置Reduce的key Reduce這個地方只有輸出的參數可以設置. 方法名字也沒有Reduce關鍵字區別於Map
        job.setOutputValueClass(LongWritable.class);//設置Reduce的value.
        
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);//表示結束了才退出,不結束不退出
    }
    /**
     * 4個泛型的意識
     * 第一個是LongWritable,固定就是這個類型,表示每一行單詞的起始位置(單位是字節)
     * 第二個是Text,表示每一行的文本內容.
     * 第三個是Text,表示單詞
     * 第四個是LongWritable,表示單詞的出現次數
     */
    public static class WordCountMapper extends Mapper<LongWritable, Text, Text    ,LongWritable>{
        Text k2 = new Text();
        LongWritable v2 = new LongWritable();
        //增加一個計數器,這個Map調用幾次就輸出對應的次數.
        int counter = 0;
        
        
        /**
         * key和value表示輸入的信息
         * 每一行文本調用一次map函數
         */
        @Override
        protected void map(LongWritable key, Text value,Mapper<LongWritable, Text, Text, LongWritable>.Context context)    
                throws IOException, InterruptedException {
            counter  = counter + 1;
            System.out.println("mapper 調用的次數:" + counter);
            //這個map方法中的Mapper的各個泛型和上面的意識是一樣的,分別代表的是k1,v1,k2,v2
            String line = value.toString();
            System.out.println(String.format("<k1,v1>的值<"+key.get()+","+line+">"));
            String[] splited = line.split("\t");
            for (String word : splited) {
                k2.set(word);
                v2.set(1);
                System.out.println(String.format("<k2,v2>的值<"+k2.toString()+","+v2.get()+">"));
                context.write(k2, v2);//通過context對象寫出去.
            }
        }
        /**
         * 這個地方的四個泛型的意思
         * 前兩個泛型是對應的Map方法的后兩個泛型.
         * Map的輸出對應的是Reduce的輸入.
         * 第一個Text是單詞
         * 第二個LongWritable是單詞對應的次數
         * 我們想輸出的也是單詞 和 次數
         * 所以第三個和第四個的類型和第一和第二個的一樣
         * 
         * 分組指的是把相同key2的value2放到一個集合中
         *
         */
        public static class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
            LongWritable v3 = new LongWritable();
            //增加一個計數器,這個Reduce調用幾次就輸出對應的次數.
            int counter = 0;
            
            /**
             * 每一個分組調用一次reduce函數
             * 過來的k2 分別是hello you me
             * 
             */
            @Override
            protected void reduce(Text key2, Iterable<LongWritable> value2Iterable,Reducer<Text, LongWritable, Text, 
                    LongWritable>.Context context)
                            throws IOException, InterruptedException {
                counter  = counter + 1;
                System.out.println("reducer 調用的次數:" + counter);
                //第一個參數是單詞,第二個是可迭代的集合. 為什么上面的LongWritable類型的對象value2變成了一個可以迭代的結合參數?
                //因為分組指的是把相同key2的value2放到一個集合中
                long sum = 0L;
                for (LongWritable value2 : value2Iterable) {
                    System.out.println(String.format("<k2,v2>的值<"+key2.toString()+","+value2.toString()+">"));
                    sum += value2.get(); //這個value2是LongWritable類型的,不能進行+= 操作,要用get()得到其對應的java基本類型.
                    //sum表示單詞k2 在整個文本中的出現次數.
                }
                v3.set(sum);
                context.write(key2, v3);
                System.out.println(String.format("<k3,v3>的值<"+key2.toString()+","+v3.get()+">"));
            }
        }
    }
}

 

通過運行Yarn集群查看Map日志得到的輸出結果: 

查看Reduce日志產看到的輸出結果:

 

//============================================================================

以下程序是之前的寫的:注釋更加詳細:

/*
 * 一個hello文件內容如下:
 *   hello        you
 *   hello        me
 */
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.output.FileOutputFormat;

public class WordCountApp {
    public static void main(String[] args) throws Exception {
        // 在main方法寫驅動程序,把Map函數和Reduce函數組織在一起.
        // 搞一個對象把Map對象和Reduce對象都放在這個對象中,我們把這個對象稱作Job
        // 兩個形參,一個是Configuration對象,一個是Job的名稱,這樣獲得了一個Job對象;
        Job job = Job.getInstance(new Configuration(),
                WordCountApp.class.getSimpleName());
        // 對這個job進行設置
        job.setJarByClass(WordCountApp.class);// 通過這個設置可以讓框架識別你寫的代碼
        
        job.setMapperClass(MyMapper.class);// 把自定義的Map類放到job中
        job.setMapOutputKeyClass(Text.class);// 定義Map的key的輸出類型,Map的輸出是<hello,2>
        job.setMapOutputValueClass(LongWritable.class);// 定義Map的value的輸出類型

        job.setReducerClass(MyReducer.class);// 把自定義的Reducer類放到job中
        job.setOutputKeyClass(Text.class);// 因為Reduce的輸出是最終的數據,Reduce的輸出是<hello,2>
        // 所以這個方法名中沒有像Map對應的放發一樣帶有Reduce,直接就是setOutputKeyClass
        job.setOutputValueClass(LongWritable.class);// 定義reduce的value輸出

        FileInputFormat.setInputPaths(job, args[0]);// 輸入指定:傳入一個job地址.
        // 這個args[0] 就是新地址,"hdfs://192.168.0.170/hello"
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 輸出指定
        // 指定輸入和輸出路徑可以通過在這里寫死的方式,也可以通過main函數參數的形式
        // 分別是args[0]和args[1]

        // 把job上傳到yarn平台上.
        job.waitForCompletion(true);
    }

    /*
     * 對於<k1,v1>而言,每一行產生一個<k1,v1>對,<k1,v1>表示<行的起始位置,行的文本內容>
     * 就本例而言map函數總共調用兩次,因為總共只有兩行.
     * 正對要統計的文本內容可以知道總共兩行,總共會調用兩次Map函數對應產生的<k1,v1>分別是<0,hello you>
     * 和第二個<k1,v1>是<10,hello me>
     */
    private static class MyMapper extends
            Mapper<LongWritable, Text, Text, LongWritable> {
        // 這個Mapper的泛型參數是<KEYIN,VALUEIN,KEYOUT,VALUEOUT> 分別對應的是k1,v1,k2,v2
        // 我們如下講的k1,v1的類型是固定的.
        // 就本例而言，map函數會被調用2次,因為總共文本文件就只有兩行.
        
        //要定義輸出的k2和v2.本案例中可以分析出<k2,v2>是對文本內容的統計<hello,1><hello,1><you,1><me,1>
        //而且<k2,v2>的內容是和<k3,v3>中的內容是一樣的.
        Text k2 = new Text();
        LongWritable v2 = new LongWritable();
        //重寫父類Mapper中的map方法
        @Override
        protected void map(LongWritable key, Text value,
                Mapper<LongWritable, Text, Text, LongWritable>.Context context)
                throws IOException, InterruptedException {
            //通過代碼或者案例分析就可以知道k1其實沒有什么用出的.
            String line = value.toString();
            String[] splited = line.split("\t");//根據制表分隔符機進行拆分.hello和me,you之間是一個制表分隔符.
            for (String word : splited) {
                k2.set(word);
                v2.set(1);
                context.write(k2, v2);
                //用context把k2,v2寫出去,框架會寫,不用我們去管.
            }
        }
    }

    private static class MyReducer extends
            Reducer<Text, LongWritable, Text, LongWritable> {
        //這個例子中的<k2,v2>和<k3,v3>中的k是一樣的,所以這里,k2當做k3了.
        LongWritable v3 = new LongWritable();
        @Override
        protected void reduce(Text k2, Iterable<LongWritable> v2s,
                Reducer<Text, LongWritable, Text, LongWritable>.Context context)
                throws IOException, InterruptedException {
            //Reduce是對上面Map中的結果進行匯總的.
            //上面拆分出來的<k2,v2>是<hello,1><hello,1><you,1><me,1>Reduce方法中就要對其進行匯總.
            long sum = 0L;
            for(LongWritable v2:v2s){
                sum = sum +v2.get();//sum是long類型,v2是LongWritable類型
                //LongWritable類型轉換成long類型用get()方法.
                //sum的值表示單詞在整個文件中出現的中次數.
            }
            v3.set(sum);
            context.write(k2,v3);
        }
    }

}

 //===============================================================================

查看日志的時候,代碼中的System.out.println()對於Java程序輸出到控制台,但是這個地方是把Java類打成Jar包,

放到集群中去通過命令執行的.

輸出通過日志查看的.

上面對應的Log Type:stdout 

stdout:stdout（Standardoutput）標准輸出

 

另外一個關於單詞計數的總結:http://www.cnblogs.com/DreamDrive/p/5494866.html