Shuffle partition 分区

默认分区：HashPartition

影响因素：key.hashcode()、NumReducerTask 

一、基础

1、目的

Reducer处理的结果按不同的条件，存储在不同的文件中

2、语法

a、自定义分区，继承Partitioner

b、分区在mapper后reducer前，因此数据类型和mapper一致

c、在driver中，job配置自定义分区和设置reducer数量

3、reducer数量

reducer数量为 1 ，分不分区都一样

reducer数量 > 1 < 分区的数量 报错

reducer数量 > 分区数量 浪费资源(reducer)

注意：分区号要从0开始，并逐一累加

二、案例

1、需求

手机号136、137、138、139开头都分别放到一个独立的4个文件中，其他开头的放到一个文件中

2、文件

1    13736230513    192.196.100.1    www.atguigu.com    2481    24681    200
2    13846544121    192.196.100.2            264    0    200
3     13956435636    192.196.100.3            132    1512    200
4     13966251146    192.168.100.1            240    0    404
5     18271575951    192.168.100.2    www.atguigu.com    1527    2106    200
6     84188413    192.168.100.3    www.atguigu.com    4116    1432    200
7     13590439668    192.168.100.4            1116    954    200
8     15910133277    192.168.100.5    www.hao123.com    3156    2936    200
9     13729199489    192.168.100.6            240    0    200
10     13630577991    192.168.100.7    www.shouhu.com    6960    690    200
11     15043685818    192.168.100.8    www.baidu.com    3659    3538    200
12     15959002129    192.168.100.9    www.atguigu.com    1938    180    500
13     13560439638    192.168.100.10            918    4938    200
14     13470253144    192.168.100.11            180    180    200
15     13682846555    192.168.100.12    www.qq.com    1938    2910    200
16     13992314666    192.168.100.13    www.gaga.com    3008    3720    200
17     13509468723    192.168.100.14    www.qinghua.com    7335    110349    404
18     18390173782    192.168.100.15    www.sogou.com    9531    2412    200
19     13975057813    192.168.100.16    www.baidu.com    11058    48243    200
20     13768778790    192.168.100.17            120    120    200
21     13568436656    192.168.100.18    www.alibaba.com    2481    24681    200
22     13568436656    192.168.100.19            1116    954    200

3、回顾自定义Hadoop序列化

a、自定义序列化类实现Writable接口

b、自定义属性

c、无参构造函数

d、get和set

e、tostring 连接使用

f、序列化

g、反序列化

注意：序列化和反序列化的顺序相同，相当于队列

三、代码

1、自定义Hadoop序列化类

package com.flow2;

import org.apache.hadoop.io.Writable;

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

public class FlowBean implements Writable {
    private long upFlow;
    private long downFlow;
    private long sunFlow;

    public FlowBean() {
    }

    public long getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }

    public long getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }

    public long getSunFlow() {
        return sunFlow;
    }

    public void setSunFlow(long sunFlow) {
        this.sunFlow = sunFlow;
    }

    public void setSum(long upFlow, long downFlow){
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sunFlow = upFlow + downFlow;
    }

    @Override
    public String toString() {
        return upFlow + "	" + downFlow + "	" + sunFlow;
    }

    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sunFlow);
    }

    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sunFlow = in.readLong();
    }
}

2、Mapper

package com.flow2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
    FlowBean v = new FlowBean();
    Text k = new Text();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 1. 行
        String line = value.toString();
        // 2. split
        String[] words = line.split("	");
        // 3.flow
        long upFlow = Long.parseLong(words[words.length -3]);
        long downFlow = Long.parseLong(words[words.length - 2]);
        v.setUpFlow(upFlow);
        v.setDownFlow(downFlow);
        // 4.key
        String phone = words[1];
        k.set(phone);
        // 5.写入
        context.write(k, v);
    }
}

3、Reducer

package com.flow2;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class FlowReducer extends Reducer<Text,FlowBean,Text,FlowBean> {
    FlowBean v = new FlowBean();
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        // 1. 累加,一个手机号有多条记录
        long sumUp = 0;
        long sumDown = 0;
        for (FlowBean value : values) {
            sumUp += value.getUpFlow();
            sumDown += value.getDownFlow();
        }
        // 2. 设置 v
        v.setSum(sumUp, sumDown);
        // 3. 写入
        context.write(key, v);
    }
}

4、Driver

package com.flow2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class FlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        args = new String[]{"E:\a\input", "E:\a\output"};
        // 1. job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        // 2. 设置jar
        job.setJarByClass(FlowDriver.class);
        // 3. 设置 mapper 和 reducer类型
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
        // 4. 设置 mapper输出 k v
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        // 5. 设置 输出 k, v
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        // 8. 设置 分区 和 Reduce 数量
        job.setPartitionerClass(PhonePartition.class);
        job.setNumReduceTasks(5);
        // 6. 设置 输入输出
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 7. 提交 job
        boolean wait = job.waitForCompletion(true);
        System.exit(wait? 0: 1);
    }
}

注意：第8步，是后面补充的

// 8. 设置 分区 和 Reduce 数量
job.setPartitionerClass(PhonePartition.class);
job.setNumReduceTasks(5);

5、Partition

package com.flow2;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class PhonePartition extends Partitioner<Text, FlowBean> {
    /**
     * 1. 分区号 从 0 开始
     * 2. partition数据类型 与 mapper的输出类型一致，partition 在 mapper后 reducer 前
     * @param text
     * @param flowBean
     * @param numPartitions
     * @return
     */
    public int getPartition(Text text, FlowBean flowBean, int numPartitions) {
        // 核心业务逻辑
        // 1. 获取手机号
        String phone = text.toString();
        // 2. 判断
        int partition;
        String substring = phone.substring(0, 3);
        if("136".equals(substring)){
            partition = 0;
        }else if("137".equals(substring)){
            partition = 1;
        }else if ("138".equals(substring)){
            partition = 2;
        }else if ("139".equals(substring)){
            partition = 3;
        }else {
            partition = 4;
        }
        return partition;
    }
}