Hadoop案例：Reduce join

朱雀 2022-11-22 00:02 177阅读 0赞

### 文章目录 ###

*   *  输入数据
     *  期望结果
     *  需求分析
     *  自定OrderProductBean
     *  Mapper类
     *  Reducer类
     *  Driver类
     *  执行结果

## 输入数据 ##

order.txt

1001	01	1
    1002	02	2
    1003	03	3
    1004	01	4
    1005	02	5
    1006	03	6

pd.txt

01	小米
    02	华为
    03	格力

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]

## 期望结果 ##

![在这里插入图片描述][20201102201829500.png_pic_center]

## 需求分析 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]

## 自定OrderProductBean ##

package com.mr.reducejoin;
    
    import org.apache.hadoop.io.Writable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    
    public class OrderProductBean implements Writable { 
        // order数据的orderid
        private int orderId;
        // order数据的数量amount
        private int amount;
        // order数据的商品id，同时也是product数据的商品id
        private int productId;
        // product 数据的商品名字
        private String productName;
        // 标记，用来标记bean对象属于order还是属于product
        private String flag;
    
        public OrderProductBean() { 
        }
    
        public OrderProductBean(int orderId, int amount, int productId, String productName, String flag) { 
            this.orderId = orderId;
            this.amount = amount;
            this.productId = productId;
            this.productName = productName;
            this.flag = flag;
        }
    
        public int getOrderId() { 
            return orderId;
        }
    
        public void setOrderId(int orderId) { 
            this.orderId = orderId;
        }
    
        public int getAmount() { 
            return amount;
        }
    
        public void setAmount(int amount) { 
            this.amount = amount;
        }
    
        public int getProductId() { 
            return productId;
        }
    
        public void setProductId(int productId) { 
            this.productId = productId;
        }
    
        public String getProductName() { 
            return productName;
        }
    
        public void setProductName(String productName) { 
            this.productName = productName;
        }
    
        public String getFlag() { 
            return flag;
        }
    
        public void setFlag(String flag) { 
            this.flag = flag;
        }
    
        @Override
        public String toString() { 
            return orderId +
                    "\t" +
                    productName + '\t' + amount
                    ;
        }
    
        @Override
        public void write(DataOutput out) throws IOException { 
            out.writeInt(orderId);
            out.writeInt(amount);
            out.writeInt(productId);
            out.writeUTF(productName);
            out.writeUTF(flag);
        }
    
        @Override
        public void readFields(DataInput in) throws IOException { 
            orderId = in.readInt();
            amount = in.readInt();
            productId = in.readInt();
            productName = in.readUTF();
            flag = in.readUTF();
        }
    }

## Mapper类 ##

package com.mr.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 ReduceJoinMapper extends Mapper<LongWritable, Text, Text, OrderProductBean> { 
    
        // 文件名字
        private String fileName;
        // bean对象，封装order和product的数据
        OrderProductBean orderProductBean = new OrderProductBean();
        // 公共字段作为key，商品的id
        Text pId = new Text();
    
        // 获取到输入文件的名字
        @Override
        protected void setup(Context context) throws IOException, InterruptedException { 
            InputSplit inputSplit = context.getInputSplit();
            FileSplit fileSplit = (FileSplit) inputSplit;
            fileName = fileSplit.getPath().getName();
        }
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            String[] split = value.toString().split("\t");
    // 判断文件名，如果包含order，表示是来自order.txt 的数据
            if (fileName.contains("order")) { 
                orderProductBean.setAmount(Integer.parseInt(split[2]));
                orderProductBean.setFlag("order");
                orderProductBean.setOrderId(Integer.parseInt(split[0]));
                orderProductBean.setProductId(Integer.parseInt(split[1]));
                // 没有的数据设置默认值
                orderProductBean.setProductName("");
    // 设置pId
                pId.set(split[1]);
            } else { 
    // 表示来自pd.txt 的数据
                orderProductBean.setProductName(split[1]);
                orderProductBean.setProductId(Integer.parseInt(split[0]));
                orderProductBean.setFlag("product");
    // 没有的数据设置默认值
                orderProductBean.setAmount(0);
                orderProductBean.setOrderId(0);
                // 设置pId
                pId.set(split[0]);
            }
    
            context.write(pId, orderProductBean);
    
        }
    }

## Reducer类 ##

package com.mr.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 ReduceJoinReducer extends Reducer<Text, OrderProductBean, OrderProductBean, NullWritable> \{

@Override
    protected void reduce(Text key, Iterable<OrderProductBean> values, Context context) throws IOException, InterruptedException {
        ArrayList<OrderProductBean> orderBeanList = new ArrayList<>();
        OrderProductBean productBean = new OrderProductBean();

// 需要分开来自order的bean对象以及来自pd 的bean对象  
for (OrderProductBean value : values) \{  
if (value.getFlag().equals(“order”)) \{  
// orderBeanList.add(value);  
OrderProductBean bean = new OrderProductBean();  
try \{  
BeanUtils.copyProperties(bean, value);  
\} catch (IllegalAccessException | InvocationTargetException e) \{  
e.printStackTrace();  
\}  
orderBeanList.add(bean);  
\} else \{  
try \{  
BeanUtils.copyProperties(productBean, value);  
\} catch (IllegalAccessException | InvocationTargetException e) \{  
e.printStackTrace();  
\}  
\}  
\}

// 迭代 orderBeanList,设置列表里面每个bean对象的productName字段  
for (OrderProductBean bean : orderBeanList) \{  
bean.setProductName(productBean.getProductName());  
context.write(bean, NullWritable.get());  
\}

}

## Driver类 ##

package com.mr.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 ReduceJoinDriver \{  
public static void main(String\[\] args) throws IOException, ClassNotFoundException, InterruptedException \{  
args = new String\[\]\{“G:\\file\\input”, “G:\\file\\ouput”\};

Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
    
    
        job.setJarByClass(ReduceJoinDriver.class);
        job.setMapperClass(ReduceJoinMapper.class);
        job.setReducerClass(ReduceJoinReducer.class);
    
    
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(OrderProductBean.class);
    
        job.setOutputKeyClass(OrderProductBean.class);
        job.setOutputValueClass(NullWritable.class);
    
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
        boolean waitForCompletion = job.waitForCompletion(true);
        System.exit(waitForCompletion ? 0 : 1);
    
    
    }

## 执行结果 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201102201751691.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw,size_16,color_FFFFFF,t_70#pic_center
[20201102201829500.png_pic_center]: https://img-blog.csdnimg.cn/20201102201829500.png#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201102201846881.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201102202035892.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjc0MzYw,size_16,color_FFFFFF,t_70#pic_center