package com.my.hadoop.mapreduce.partition;


import java.util.HashMap;
import java.util.Map;

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.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ParCount {

    public static class ParMap extends Mapper<LongWritable, Text, Text, InfoBean>{
        private Text key = new Text();
        @Override
        public void map(LongWritable key, Text value, Context context) throws java.io.IOException ,InterruptedException {
            String[] fields = value.toString().split("\t");
            String telNo = fields[1];
            long upPayLoad = Long.parseLong(fields[8]);
            long downPayLoad = Long.parseLong(fields[9]);
            InfoBean bean = new InfoBean(telNo, upPayLoad, downPayLoad);
            this.key.set(telNo);
            context.write(this.key, bean);
        }
    }
    
    public static class ParReduce extends Reducer<Text, InfoBean, Text, InfoBean>{
        @Override
        public void reduce(Text key, java.lang.Iterable<InfoBean> value, org.apache.hadoop.mapreduce.Reducer<Text,InfoBean,Text,InfoBean>.Context context) throws java.io.IOException ,InterruptedException {
            long up_sum = 0;
            long down_sum = 0;
            for (InfoBean bean : value) {
                up_sum += bean.getUpPayLoad();
                down_sum += bean.getDownPayLoad();
            }
            InfoBean bean = new InfoBean("", up_sum, down_sum);
            context.write(key, bean);
        }
    }
    
    /**
     * 分区，参数为Map的输出
     * @author yao
     *
     */
    public static class MyPar extends Partitioner<Text, InfoBean>{

        private static Map<String, Integer> parFlag = new HashMap<String, Integer>();
        static {
            parFlag.put("135", 1);
            parFlag.put("136", 1);
            parFlag.put("137", 1);
            parFlag.put("138", 1);
            parFlag.put("139", 1);
            parFlag.put("150", 2);
            parFlag.put("159", 2);
            parFlag.put("182", 3);
            parFlag.put("183", 3);
        }
        
        @Override
        public int getPartition(Text key, InfoBean value, int arg2) {
            String telNo = key.toString().substring(0, 3);
            Integer code = parFlag.get(telNo);
            if (code == null) {
                code = 0;
            }
            
            return code;
        }
        
    }
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, ParCount.class.getSimpleName());;
        job.setJarByClass(ParCount.class);
        
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        job.setMapperClass(ParMap.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(InfoBean.class);
        
        job.setPartitionerClass(MyPar.class);                                        //指定自定义的分区类
        job.setNumReduceTasks(4);　　　　　　　　　　　　　　　　　　 //需要根据分区的数量设置Reducer数量，多了会出现空文件，少了会报错
        
        job.setReducerClass(ParReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(InfoBean.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
        
    }

}

hadoop2.2.0 MapReduce分区