建站记录查询云南政务服务网

MapReduce程序之Join案例

案例

现有数据文件  order.txt、  user.txt

用户文件中存储用户数据  
订单文件中存储订单数据
1个用户可对应多条订单数据
现要求将数据进行汇总，且以订单数据为基准，拼接完整数据

order.txt

 user.txt

思路：

两个文件中关联字段为uid，1个用户对应多条订单数据，以订单数据为准，
即拿到订单数据的集合，循环它，拼接上用户数据输出即可

思考：mapreduce程序设计

reduce 最终需要输出：完整拼接数据，最好进行排序，所以以完整拼接数据（javabean）为key，value为nullwritable
map  输入：  long（k）、单行数据（v） ;

有个问题：map任务中重写的map方法只有一个，但是我们的两个文件的数据结构是截然不同的
但我们经过分析发现，最终需要的数据是两个文件的结合而成，我们可以自定义一个pojo，使得同时
包含上述两种文件的数据字段，这样就能处理两种文件了，为了在reduce中区分数据，我们还需要多定义一个数据区分
字段 dataType 用以区分 用户数据还是订单数据;为了map输出 reduce输入  网络中传输  还需要实现属性序列化规则

但是程序如何知道读进来的数据是哪个文件的呢？
这里我们可以使用文件名来进行判断 因为在一个map任务中，fileName是不变的
而且读取数据之后 进行处理 汇总，我们找到两个数据文件字段的共同字段（关联字段）uid ，作为key

MR程序及主程序代码如下：

package cn.doit19.hadoop.review.join;import cn.doit19.hadoop.bean.JoinBean;
import org.apache.commons.beanutils.BeanUtils;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;/*** @author:tom* @Date:Created in 21:43 2020/11/17*/
@SuppressWarnings("all")
public class JoinApp {static class JoinMap extends Mapper<LongWritable, Text, Text, JoinBean> {String fileName = null;Text k = new Text();JoinBean v = new JoinBean();@Overrideprotected void setup(Context context) throws IOException, InterruptedException {FileSplit fs = (FileSplit) context.getInputSplit();fileName = fs.getPath().getName();}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString();if (fileName.startsWith("user")) {//用户数据String[] strs = line.split(",");//u001,senge,18,male,angelababyk.set(strs[0]);v.set(strs[0], strs[1], Integer.parseInt(strs[2]), strs[3], strs[4], "-1", "user");} else {//订单数据//order011, u001String[] strs = line.split(",");k.set(strs[1]);v.set("-1", "-1", -1, "-1", "-1", strs[0], "order");}context.write(k, v);}}static class JoinReduce extends Reducer<Text, JoinBean, JoinBean, NullWritable> {List<JoinBean> list = new ArrayList<>();JoinBean v = new JoinBean();@Overrideprotected void reduce(Text key, Iterable<JoinBean> values, Context context) throws IOException, InterruptedException {try {for (JoinBean value : values) {if ("user".equals(value.getDataType())) {BeanUtils.copyProperties(v, value);} else {//订单数据JoinBean order = new JoinBean();BeanUtils.copyProperties(order, value);list.add(order);}}} catch (Exception e) {e.printStackTrace();}for (JoinBean joinBean : list) {joinBean.set(v.getUid(), v.getName(), v.getAge(), v.getName(), v.getFriends(), joinBean.getOid(), "aaa");context.write(joinBean, NullWritable.get());}}}public static void main(String[] args) throws Exception {//初始化配置对象Configuration conf = new Configuration();//创建job对象Job job = Job.getInstance(conf);//设置map task 类job.setMapperClass(JoinMap.class);//设置reduce task 类job.setReducerClass(JoinReduce.class);//设置map输出类型  kvjob.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(JoinBean.class);//设置reduce 最终输出类型  kvjob.setOutputKeyClass(JoinBean.class);job.setOutputValueClass(NullWritable.class);//设置reduce 数量
//        job.setNumReduceTasks(2);//设置输入路径FileInputFormat.setInputPaths(job, new Path("E:\\MR\\In\\join"));//设置输出路径FileOutputFormat.setOutputPath(job, new Path("E:\\MR\\out\\join"));//提交任务boolean s = job.waitForCompletion(true);}
}

JoinBean类代码：

package cn.doit19.hadoop.bean;import org.apache.hadoop.io.Writable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;/*** @author:tom* @Date:Created in 22:12 2020/11/17*/
public class JoinBean implements Writable {private String uid;private String name;private int age;private String gender;private String friends;private String oid;private String dataType;public String getUid() {return uid;}public void setUid(String uid) {this.uid = uid;}public String getName() {return name;}public void setName(String name) {this.name = name;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}public String getGender() {return gender;}public void setGender(String gender) {this.gender = gender;}public String getFriends() {return friends;}public void setFriends(String friends) {this.friends = friends;}public String getOid() {return oid;}public void setOid(String oid) {this.oid = oid;}public String getDataType() {return dataType;}public void setDataType(String dataType) {this.dataType = dataType;}@Overridepublic String toString() {return "JoinBean{" +"uid='" + uid + '\'' +", name='" + name + '\'' +", age=" + age +", gender='" + gender + '\'' +", friends='" + friends + '\'' +", oid='" + oid + '\'' +", dataType='" + dataType + '\'' +'}';}public void set(String uid, String name, int age, String gender, String friends, String oid, String dataType) {this.uid = uid;this.name = name;this.age = age;this.gender = gender;this.friends = friends;this.oid = oid;this.dataType = dataType;}@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeUTF(uid);dataOutput.writeUTF(name);dataOutput.writeInt(age);dataOutput.writeUTF(gender);dataOutput.writeUTF(friends);dataOutput.writeUTF(oid);dataOutput.writeUTF(dataType);}@Overridepublic void readFields(DataInput dataInput) throws IOException {uid = dataInput.readUTF();name = dataInput.readUTF();age = dataInput.readInt();gender = dataInput.readUTF();friends = dataInput.readUTF();oid = dataInput.readUTF();dataType = dataInput.readUTF();}
}

输出结果：

 

实现要点注意：

1.两个文件的数据结构不一样，读取进来，map处理方式肯定不一样，但不可能写两个map，最后输出结果又必须结合两个数据结构，所以以文件名区分数据结构；构造统一结合数据结构JoinBean。

文件名获取方法：XXXMap类 重写父类的setup方法，使用其上下文对象context

        @Overrideprotected void setup(Context context) throws IOException, InterruptedException {FileSplit fs = (FileSplit) context.getInputSplit();fileName = fs.getPath().getName();}

2.由于map 端需要输出到磁盘    reduce需要从磁盘中读入数据   均涉及到对象的序列化，而jdk自带的序列化数据冗余，所以选择使用hadoop的序列化规则，即pojo类实现Writable接口，重写write和read方法

【pojo类、domain、javabean均指joinbean这样的类】

3.设置JoinBean的属性时，由于是订单数据，则用户的相关属性值就没有，但也不能设置为NULL，因为有序列化规则在，如果设置为null，会在序列化时就报空指针异常

4.

这里不能使用在外界定义joinbean，每次重新赋值，添加进list的形式；这样最终添加进去的是同一个joinbean

更多学习、面试资料尽在微信公众号：Hadoop大数据开发