MapReducer自定义bean-排序-分组和shuffle的过程

package cn.itcast.hadoop.mr.flowsum;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;

/**
 * FlowBean 是我们自定义的数据类型，要在hadoop的各个节点之间传输，应遵循hadoop的序列化机制
 * 就必须实现hadoop相应的序列化接口
 * */

public class FlowBean implements Writable {

    private String phoneNB;
    private long up_flow;
    private long d_flow;
    private long s_flow;

    // 在反序列化时，反射机制需要调用空参构造函数
    public FlowBean(){}

    public FlowBean(String phoneNB, long up_flow, long d_flow) {
        super();
        this.phOneNB= phoneNB;
        this.up_flow = up_flow;
        this.d_flow = d_flow;
    }

    public String getPhoneNB() {
        return phoneNB;
    }

    public void setPhoneNB(String phoneNB) {
        this.phOneNB= phoneNB;
    }

    public long getUp_flow() {
        return up_flow;
    }

    public void setUp_flow(long up_flow) {
        this.up_flow = up_flow;
    }

    public long getD_flow() {
        return d_flow;
    }

    public void setD_flow(long d_flow) {
        this.d_flow = d_flow;
    }

    /**
     * @return the s_flow
     */
    public long getS_flow() {
        return s_flow;
    }

    /**
     * @param s_flow the s_flow to set
     */
    public void setS_flow(long s_flow) {
        this.s_flow = s_flow;
    }


    @Override
    public String toString() {

        return ""+ phoneNB + "\t" + up_flow + "\t" + d_flow;
    }


    // 将对象数据序列化到流中
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(phoneNB);
        out.writeLong(up_flow);
        out.writeLong(d_flow);
        out.writeLong(s_flow);

    }

    // 从数据流中泛序列出对象的数据
    // 从数据流中读出对象字段时，必须跟序列化时的顺序保持一致
    @Override
    public void readFields(DataInput in) throws IOException {

        phoneNB = in.readUTF();
        up_flow = in.readLong();
        d_flow = in.readLong();
        s_flow = in.readLong();
    }
}

package cn.itcast.hadoop.mr.flowsum;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
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;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;


public class FlowSumRunner extends Configured implements Tool {

    public static class FlowSumMapper extends Mapper {
        //拿到日志中的数据，切分各个字段
        @Override
        protected void map(LongWritable key, Text value, Context context)
         throws IOException, InterruptedException{
　　　　　　　　//拿一行数据
            String line = value.toString();

            String[] fileds = StringUtils.split(line, "\t");

            String phoneNB = fileds[1];
            long up_flow = Long.parseLong(fileds[7]);
            long d_flow = Long.parseLong(fileds[8]);
　　　　　　　　//封装数据为kv并输出
            context.write(new Text(phoneNB), new FlowBean(phoneNB,up_flow, d_flow));
        }
    }


    public static class FlowSumReducer extends Reducer {

　　　　　　//框架每传递一组数据<1387788654,{flowbean,flowbean,flowbean,flowbean.....}>调用一次我们的reduce方法
　　　　　　//reduce中的业务逻辑就是遍历values，然后进行累加求和再输出

        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

            long up_flow_counter = 0;
            long d_flow_counter= 0;

            for (FlowBean value : values) {
                up_flow_counter += value.getUp_flow();
                d_flow_counter += value.getD_flow();
            }
            context.write(key, new FlowBean(key.toString(),up_flow_counter, d_flow_counter));
        }
    }

    @Override
    public int run(String[] args) throws Exception {

        Configuration conf = new Configuration(); 

        Job job = Job.getInstance(conf);

        job.setJarByClass(FlowSumRunner.class);

        job.setMapperClass(FlowSumMapper.class);
        job.setReducerClass(FlowSumReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));  //读取命令行参数
        FileOutputFormat.setOutputPath(job, new Path(args[1]));//读取命令行参数

        return job.waitForCompletion(true)?0:1;
    }

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

    //hadoop jar flow.jar cn.itcast.hadoop.mr.flowsum.FlowSumRunner /flow/data /flow/output

        int res = ToolRunner.run(new Configuration(), new FlowSumRunner(), args);
        System.exit(res);
    }
}

package cn.itcast.hadoop.mr.flowsum;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class FlowBean implements WritableComparable {

    ...
    ...
    ...

    @Override
    public int compareTo(FlowBean o) {
        return this.getUp_flow() > o.getUp_flow() ? -1:1;
    }
}//如果对象本身的id大于传入的对象id，

//返回值是正数，就是升序排序

//返回值是负数，就是降序排序

package cn.itcast.hadoop.mr.flowsort;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;

import cn.itcast.hadoop.mr.flowsum.FlowBean;

public class SortMR {

    // NullWritable 如果什么也不想输出的话  使用NumWritable
    public static class SortMapper extends Mapper {

        @Override
        protected void map(LongWritable key, Text value, Context context )
        throws IOException, InterruptedException{
            String line = value.toString();

            String[] fields = StringUtils.split(line,"\t");

            String phoneNB = fields[0];
            long up_flow = Long.parseLong(fields[1]);
            long d_flow = Long.parseLong(fields[2]);


            context.write(new FlowBean(phoneNB, up_flow, d_flow), NullWritable.get());
        }
    }

    public static class SortReducer extends Reducer {

        @Override
        protected void reduce(FlowBean key, Iterable values, Context context)
                 throws IOException, InterruptedException

        {
            context.write(key, NullWritable.get());
        }
    }

    public static void main(String[] args) 
            throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration(); 

        Job job = Job.getInstance(conf);

        job.setJarByClass(SortMR.class);

        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReducer.class);

        job.setMapOutputKeyClass(FlowBean.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(FlowBean.class);
        job.setOutputValueClass(NullWritable.class);

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


        int result = job.waitForCompletion(true)?0:1;

        System.exit(result);
    }

}

13600217502 186852  200
13560439658 4938    200
84138413    4116    1432
13922314466 3008    3720
13726230503 2481    24681
13926435656 1512    200
13560439658 954 200
13480253104 180 200
13823070001 180 200
13760778710 120 200
13502468823 102 7335
13925057413 63  11058
15013685858 27  3659
15920133257 20  3156
18320173382 18  9531
18211575961 12  1527
13602846565 12  1938
13660577991 9   6960
15989002119 3   1938
13719199419 0   200
13826544101 0   200
13926251106 0   200

package cn.itcast.hadoop.mr.areapartition;

import java.util.HashMap;
import org.apache.hadoop.mapreduce.Partitioner;

public  class AreaPartitioner extends Partitioner {

    private static  HashMap areaMap = new HashMap<>();

    static {        //暂且先静态写死分组
        areaMap.put("135", 0);
        areaMap.put("136", 1);
        areaMap.put("137", 2);
        areaMap.put("138", 3);
        areaMap.put("139", 4);
    }

    @Override
    public int getPartition(KEY key, VALUE value, int numPartitions) {

        // 从key中拿到手机号，查询手机归属地字典， 不同省份返回不同的编组
        Integer code = areaMap.get(key.toString().substring(0,3));
        int areaCoder = code  ==  null ? 5:code;

        return areaCoder;
    }

}

job.setPartitionerClass(AreaPartitioner.class);
job.setNumReduceTasks(6); //然后设置启动Reducer的数量

// 这里需要注意的是  这里的数量必须大于等于 你设置的数据分组的数量  不然会进行报错  
//  如果不设置 默认为1  就不会分组， 所有数据就只有一个文件
// 如果设置的多了  多的文件里面不会有数据

hadoop jar flowarea.jar cn.itcast.hadoop.mr.areapartition.FlowSumArea /flow/data  /flow/areaoutput

[hadoop@hadoop1 ~]$ hadoop fs -ls /flow/areaoutput
Found 7 items
-rw-r--r--   1 hadoop supergroup          0 2018-03-18 13:57 /flow/areaoutput/_SUCCESS
-rw-r--r--   1 hadoop supergroup         66 2018-03-18 13:56 /flow/areaoutput/part-r-00000
-rw-r--r--   1 hadoop supergroup         98 2018-03-18 13:56 /flow/areaoutput/part-r-00001
-rw-r--r--   1 hadoop supergroup         97 2018-03-18 13:57 /flow/areaoutput/part-r-00002
-rw-r--r--   1 hadoop supergroup         62 2018-03-18 13:57 /flow/areaoutput/part-r-00003
-rw-r--r--   1 hadoop supergroup        130 2018-03-18 13:57 /flow/areaoutput/part-r-00004
-rw-r--r--   1 hadoop supergroup        219 2018-03-18 13:57 /flow/areaoutput/part-r-00005

[hadoop@hadoop1 ~]$ hadoop fs -cat /flow/areaoutput/part-r-00000
13502468823 13502468823 102 7335
13560439658 13560439658 5892    400