MR案例：链式ChainMapper

类似于Linux管道重定向机制，前一个Map的输出直接作为下一个Map的输入，形成一个流水线。设想这样一个场景：在Map阶段，数据经过mapper01和mapper02处理；在Reduce阶段，数据经过sort和shuffle后，交给对应的reducer处理。reducer处理后并没有直接写入到Hdfs, 而是交给了另一个mapper03处理，它产生的最终结果写到hdfs输出目录中。

注意：对任意MR作业，Map和Reduce阶段可以有无限个Mapper，但reduer只能有一个。  

package chain;

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.io.VLongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Chain {

    /**
     *    手机    5000    * 需求： 
     *    电脑    2000    * 在第一个Mapper1里面过滤大于10000的数据 
     *    衣服    300　　　* 第二个Mapper2里面过滤掉大于100-10000的数据 
     *    鞋子    1200    * Reduce里面进行分类汇总并输出 
     *    裙子    434     * Reduce后的Mapper3里过滤掉商品名长度大于3的数据 
     *    手套    12      *
     *    图书    12510   * 
     *    小商品    5　　　* 结果： 
     *    小商品    3     * 手套 12 
     *    订餐      2     * 订餐 2 
     */

    public static void main(String[] args) throws Exception {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(Chain.class);

        /**
         * 配置mapper1
         * 注意此处带参数的构造函数：new Configuration(false)
         */
        Configuration map1Conf = new Configuration(false);
        ChainMapper.addMapper(job,         //主作业
                Mapper1.class,             //待加入的map class
                LongWritable.class,        //待加入map class的输入key类型
                Text.class,                //待加入map class的输入value类型 
                Text.class,                //待加入map class的输出key类型
                VLongWritable.class,       //待加入map class的输出value类型
                map1Conf);                 //待加入map class的配置信息

        //配置mapper2
        ChainMapper.addMapper(job, Mapper2.class, Text.class, VLongWritable.class, Text.class, VLongWritable.class, new Configuration(false));

        /**
         * 配置Reducer
         * 注意此处使用的是setReducer()方法
         */
        ChainReducer.setReducer(job, Reducer_Only.class, Text.class, VLongWritable.class, Text.class, VLongWritable.class, new Configuration(false));

        //配置mapper3
        ChainReducer.addMapper(job, Mapper3.class, Text.class, VLongWritable.class, Text.class, VLongWritable.class, new Configuration(false));

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

        job.waitForCompletion(true);
    }

    //Mapper1
    public static class Mapper1 extends Mapper<LongWritable, Text, Text, VLongWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            
            /**
             * Hadoop中默认的输入格式 TextOutputFormat 只支持UTF-8格式
             * 所以解决GBK中文输出乱码问题的方法是：
             * 1. 先将输入的Text类型的value转换为字节数组
             * 2. 然后使用String的构造器String(byte[] bytes, int offset, int length, Charset charset)
             * 3. 通过使用指定的charset解码指定的byte子数组，构造一个新的String
             */
            String line=new String(value.getBytes(),0,value.getLength(),"GBK");
            String[] splited = line.split(" ");

            //过滤大于10000的数据
            if(Integer.parseInt(splited[1])<10000L){
                context.write(new Text(splited[0]), new VLongWritable(Long.parseLong(splited[1])));
            }
        }
    }

    //Mapper2
    public static class Mapper2 extends Mapper<Text, VLongWritable, Text, VLongWritable>{
        @Override
        protected void map(Text key, VLongWritable value, Context context)
                throws IOException, InterruptedException {

            //过滤100-10000间的数据
            if(value.get()<100L){
                context.write(key, value);
            }
        }
    }

    //Reducer
    public static class Reducer_Only extends Reducer<Text, VLongWritable, Text, VLongWritable>{
        @Override
        protected void reduce(Text key, Iterable<VLongWritable> v2s, Context context)
                throws IOException, InterruptedException {

            long sumLong=0L;

            for(VLongWritable vLongWritable : v2s){
                sumLong += vLongWritable.get();

                context.write(key, new VLongWritable(sumLong));
            }
        }
    }

    //Mapper3
    public static class Mapper3 extends Mapper<Text, VLongWritable, Text, VLongWritable>{
        @Override
        protected void map(Text key, VLongWritable value, Context context)
                throws IOException, InterruptedException {

            String line=new String(key.getBytes(),0,key.getLength(),"GBK");
            
            //过滤商品名称长度大于3            
            if(line.length()<3){
                context.write(key, value);
            }
        }
    }
}