hadoop3.x MapReduce案例-Partition分区的实现

作者：马育民 • • 阅读：10198

实现分区

自定义 Partitioner

  1. 创建类，继承 org.apache.hadoop.mapreduce.Partitioner<KEY, VALUE> 类
    提示：<KEY, VALUE> 与 mapper输出的泛型相同

  2. 重写 getPartition() 方法

设置 Partitioner

在 job 中设置 Partitioner

job.setPartitionerClass();

设置 reduce task 数量

默认 reduce task 数量是 1，不会往多个文件中写数据

有多少个结果文件，就设置多少个数量：

job.setNumReduceTask(n);

案例

上接：hadoop3.x MapReduce序列化案例-统计充电桩充电量、充电时长、充电次数 案例

充电桩编号,充电量,开始时间,结束时间

1,100,2020-01-01 09:30:50,2020-01-01 09:40:50
2,200,2020-01-11 10:40:50,2020-01-11 11:05:50
3,150,2020-01-01 11:20:50,2020-01-01 11:32:50
1,300,2020-01-03 14:30:50,2020-01-03 15:20:10
1,50,2020-02-02 12:30:50,2020-02-02 12:35:50
2,500,2020-02-02 15:30:30,2020-02-02 17:20:30
3,150,2020-02-01 11:20:50,2020-02-01 11:32:50
3,150,2020-02-03 11:20:50,2020-02-03 11:32:50
1,100,2020-03-01 09:30:50,2020-03-01 09:40:50
1,100,2020-03-05 09:30:50,2020-03-05 09:40:50
2,500,2020-03-12 15:30:30,2020-03-12 17:30:30
2,500,2020-03-22 15:30:30,2020-03-22 17:30:30
3,400,2020-03-02 17:30:20,2020-03-02 18:20:20
4,100,2020-04-01 11:20:50,2020-04-01 11:25:50
4,50,2020-04-21 11:20:50,2020-04-21 11:28:50
5,150,2020-04-11 11:20:50,2020-04-11 11:32:50
6,150,2020-04-15 11:20:50,2020-04-15 11:40:50

按照 充电月份 分成不同的文件

代码实现

Cdz

需要增加 startTime、endTime 属性和 setter、getter 方法

package partition;

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

import org.apache.hadoop.io.Writable;

public class Cdz implements Writable {

    /**
     * 充电桩编号
     */
    private String id;
    /**
     * 充电次数
     */
    private int num;
    /**
     * 用电量
     */
    private int eleNum;
    /**
     * 充电时长，单位：分钟
     */
    private int time;
    /**
     * 充电开始时间
     */
    private String startTime;
    /**
     * 充电结束时间
     */
    private String endTime;

    //序列化
    @Override
    public void write(DataOutput out) throws IOException {
        //写的顺序与读的顺序相同，否则读写错乱
        out.writeUTF(id);
        out.writeInt(num);
        out.writeInt(eleNum);
        out.writeInt(time);
    }

    //反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        //写的顺序与读的顺序必须相同，否则读写错乱
        id=in.readUTF();
        num=in.readInt();
        eleNum=in.readInt();
        time=in.readInt();
    }

    //保存文件的格式
    @Override
    public String toString() {

        return id+","+num+","+eleNum+","+time;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public int getTime() {
        return time;
    }

    public void setTime(int time) {
        this.time = time;
    }

    public int getEleNum() {
        return eleNum;
    }

    public void setEleNum(int eleNum) {
        this.eleNum = eleNum;
    }

    public String getStartTime() {
        return startTime;
    }

    public void setStartTime(String startTime) {
        this.startTime = startTime;
    }

    public String getEndTime() {
        return endTime;
    }

    public void setEndTime(String endTime) {
        this.endTime = endTime;
    }

}

MyPartition

自定义分区实现类

package partition;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class MyPartition extends Partitioner<Text, Cdz> {

    @Override
    public int getPartition(Text key, Cdz value, int numPartitions) {
        String startTime=value.getStartTime();
        String month=startTime.substring(5,7);
        int res=0;
        boolean b=StringUtils.isNumeric(month);
        if(b){    
            //文件从0开始编号，所以月份要减1
            res=Integer.parseInt( month ) - 1;
        }
        return res;
    }

}

自定义map类

给 Cdz 对象传入 startTime 和 endTime

package partition;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

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

public class MyMapper extends Mapper<LongWritable, Text, Text, Cdz> {

    private Text idText=new Text();
    private Cdz cdz=new Cdz();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Cdz>.Context context)
            throws IOException, InterruptedException {

        //读取一行
        String line=value.toString();
        //根据 , 拆分
        String[] array=line.split(",");

        //拆分数据
        String id=array[0];
        int eleNum=Integer.parseInt(array[1]);
        String startTime=array[2];
        String endTime=array[3];

        SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            //计算充电时长
            Date startDate = sdf.parse(startTime);
            Date endDate=sdf.parse(endTime);
            int time=(int)(endDate.getTime()-startDate.getTime())/1000/60;

            //封装
            cdz.setId(id);
            cdz.setNum(1);
            cdz.setEleNum(eleNum);
            cdz.setTime(time);
            cdz.setStartTime(startTime);
            cdz.setEndTime(endTime);

            idText.set(id);

            //写数据
            context.write(idText, cdz);
        } catch (ParseException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }



    }



}

自定义 reduce 类

与上一案例相同

package partition;

import java.io.IOException;

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

public class MyReducer extends Reducer<Text, Cdz, Text, Cdz>{
    private Cdz cdz=new Cdz();
    @Override
    protected void reduce(Text arg0, Iterable<Cdz> arg1, Reducer<Text, Cdz, Text, Cdz>.Context arg2)
            throws IOException, InterruptedException {
        int numTotal=0;//充电总次数
        int eleNumTotal=0;//充电总量
        int timeTotal=0;//充电总时长

        //遍历，累加
        for(Cdz item:arg1){
            numTotal ++;
            eleNumTotal+= item.getEleNum();
            timeTotal += item.getTime();
        }
        //封装
        cdz.setId(arg0.toString());
        cdz.setNum(numTotal);
        cdz.setEleNum(eleNumTotal);
        cdz.setTime(timeTotal);
        //写数据
        arg2.write(arg0, cdz);
    }

}

driver 类

package partition;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Main2 extends Configured implements Tool{
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        //通过ToolRunner类 执行 覆盖的run()方法
        int res=ToolRunner.run(config, new Main2(), args);
        //退出客户端
        System.exit(res);
    }


    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(),"mywordcount2");
        //设置jar加载路径，不加此代码有时可能会报错
        job.setJarByClass(Main2.class);
        //设置map和reduce类
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        //设置map输出 key/value 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Cdz.class);
        //设置最终输出 key/value 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Cdz.class);

        //指定自定义的Partitioner
        job.setPartitionerClass(MyPartition.class);
        //设置数量，否则分区无效
        job.setNumReduceTasks(12);

        //根据输入参数 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //提交
//        job.submit();
        boolean result = job.waitForCompletion(true);
        return result?1:0;
    }
}
关键

指定自定义的Partitioner：

job.setPartitionerClass(MyPartition.class);

设置 reduce task 数量，否则分区无效：

job.setNumReduceTasks(12);

原文出处：/show_1IX2CpCC2iHO.html

吉ICP备15003043号