Hadoop3教程(十七):MapReduce之ReduceJoin案例分析

(113)ReduceJoin案例需求分析

现在有两个文件:

  • orders.txt,存放的是订单ID、产品ID、产品数量
  • pd.txt,这是一个产品码表,存放的是产品ID、产品中文名;

现在是想通过join,来实现这么一个预期输出,即订单ID、产品中文名、产品数量。

以上是本次案例需求。

简单思考一下思路。我们需要将关联条件作为Map输出的key,将两表满足Join条件的数据以及数据所来源的文件信息,发往同一个ReduceTask,在Reduce中进行数据的串联。

具体该怎么做呢?

Map中在处理的时候,需要获取输入的文件内容和文件名(这个是可以在切片的时候获取的),然后不同文件分别做不同处理,处理完成后封装bean对象输出。

注意,Map在输出的时候,需要以产品ID作为key,只有这样做,才能将相同产品ID的orders.txt记录和pd.txt记录,放在同一个reduceTask里,进而实现最终的替换。value的话,选择订单ID、订单数量、文件名。这里传入文件名的原因是Reduce阶段需要根据不同文件名实现不同处理,所以一定得需要传一个文件名进来。

另外提一句,封装bean对象的时候,需要把两个文件里的所有字段合起来作为一个bean对象,这样子,orders文件的数据可以用这个bean对象,pd.txt里的数据也可以用这个bean对象。相当于做一个大宽表。

reduce阶段就很简单了,相同产品ID的orders.txt记录和pd.txt记录,被放在同一个reduceTask里,可以把来自orders的bean放在一个集合里,来自pd的bean放在一个集合里,然后遍历set覆盖就可以。

(114)ReduceJoin案例代码实操 - TableBean

首先需要定义一个Bean对象,用来序列化两个输入文件的数据,我们命名为TableBean。

package com.atguigu.mapreduce.reducejoin;

import org.apache.hadoop.io.Writable;

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

public class TableBean implements Writable {

    private String id; //订单id
    private String pid; //产品id
    private int amount; //产品数量
    private String pname; //产品名称
    private String flag; //判断是order表还是pd表的标志字段

    public TableBean() {
    }

    public String getId() {
        return id;
    }

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

    public String getPid() {
        return pid;
    }

    public void setPid(String pid) {
        this.pid = pid;
    }

    public int getAmount() {
        return amount;
    }

    public void setAmount(int amount) {
        this.amount = amount;
    }

    public String getPname() {
        return pname;
    }

    public void setPname(String pname) {
        this.pname = pname;
    }

    public String getFlag() {
        return flag;
    }

    public void setFlag(String flag) {
        this.flag = flag;
    }

    @Override
    public String toString() {
        return id + "t" + pname + "t" + amount;
    }

    // 序列化方法
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(id);
        out.writeUTF(pid);
        out.writeInt(amount);
        out.writeUTF(pname);
        out.writeUTF(flag);
    }

    // 反序列化方法
    // 注意,序列化的顺序必须要跟反序列化的顺序一致
    @Override
    public void readFields(DataInput in) throws IOException {
        this.id = in.readUTF();
        this.pid = in.readUTF();
        this.amount = in.readInt();
        this.pname = in.readUTF();
        this.flag = in.readUTF();
    }
}

注意,序列化的顺序必须要跟反序列化的顺序一致。

(115)ReduceJoin案例代码实操 - TableMapper

TableMapper的主要作用,就是将输入的数据,划分成指定的KV对,以供Reduce阶段使用。

命名为TableMapper,获取文件名称的代码也包含在这里。

package com.atguigu.mapreduce.reducejoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class TableMapper extends Mapper<LongWritable,Text,Text,TableBean> {

    private String filename;
    private Text outK = new Text();
    private TableBean outV = new TableBean();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //获取对应文件名称
        InputSplit split = context.getInputSplit();
        FileSplit fileSplit = (FileSplit) split;
        filename = fileSplit.getPath().getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //获取一行
        String line = value.toString();

        //判断是哪个文件,然后针对文件进行不同的操作
        if(filename.contains("order")){  //订单表的处理
            String[] split = line.split("t");
            //封装outK
            outK.set(split[1]);
            //封装outV
            outV.setId(split[0]);
            outV.setPid(split[1]);
            outV.setAmount(Integer.parseInt(split[2]));
            outV.setPname("");
            outV.setFlag("order");
        }else {                             //商品表的处理
            String[] split = line.split("t");
            //封装outK
            outK.set(split[0]);
            //封装outV
            outV.setId("");
            outV.setPid(split[0]);
            outV.setAmount(0);
            outV.setPname(split[1]);
            outV.setFlag("pd");
        }

        //写出KV
        context.write(outK,outV);
    }
}

(116)ReduceJoin案例代码实操 - Reducer及Driver

主要是编写Reduce部分。Mapper之后,一组相同的key的数据会进入一个ReduceTask,接下来需要编写自定义逻辑,让Reduce可以实现关联后输出。

需要创建两个集合,每个集合接收不同文件,一个接收order文件数据,另一个接收码表数据。然后循环遍历order集合,把码表集合里的值set进去。

新建TableReducer:

package com.atguigu.mapreduce.reducejoin;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

public class TableReducer extends Reducer<Text,TableBean,TableBean, NullWritable> {

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

        ArrayList<TableBean> orderBeans = new ArrayList<>();
        TableBean pdBean = new TableBean();

        for (TableBean value : values) {

            //判断数据来自哪个表
            if("order".equals(value.getFlag())){   //订单表

			  //创建一个临时TableBean对象接收value
                TableBean tmpOrderBean = new TableBean();

                try {
                    BeanUtils.copyProperties(tmpOrderBean,value);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }

			  //将临时TableBean对象添加到集合orderBeans
                orderBeans.add(tmpOrderBean);
            }else {                                    //商品表
                try {
                    BeanUtils.copyProperties(pdBean,value);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            }
        }

        //遍历集合orderBeans,替换掉每个orderBean的pid为pname,然后写出
        for (TableBean orderBean : orderBeans) {

            orderBean.setPname(pdBean.getPname());

		   //写出修改后的orderBean对象
            context.write(orderBean,NullWritable.get());
        }
    }
}

根据教程上说的,这里只有一个地方需要注意,但是用处也不是很大,就是 集合在add value的时候,不能直接add 传进来的value,而是需要重新new一个TableBean,将value值赋值给这个新的TableBean,最后add这个新的TableBean。

这么做的原因是,传进来的values,其实是一个Iterable<TableBean> ,不是传统意义上的迭代器,可以简单理解成,Iterable<TableBean> 里的每个value用的是同一个内存地址,每次读取出value就总是赋给那个内存地址,所以不能直接add value,否则add 一百次,也只会记住最后一次add的那个value。

这似乎是Hadoop为了避免因创建过多实例引起资源浪费,而做的优化。

没有测过,做简单了解吧。

最后在驱动类里注册:

package com.atguigu.mapreduce.reducejoin;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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 TableDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());

        job.setJarByClass(TableDriver.class);
        job.setMapperClass(TableMapper.class);
        job.setReducerClass(TableReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(TableBean.class);

        job.setOutputKeyClass(TableBean.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path("D:\input"));
        FileOutputFormat.setOutputPath(job, new Path("D:\output"));

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

大功告成

参考文献

  1. 【尚硅谷大数据Hadoop教程,hadoop3.x搭建到集群调优,百万播放】