Hadoop入门（十六）Mapreduce的单表关联程序

ゞ浴缸里的玫瑰 2023-06-07 08:11 44阅读 0赞

"单表关联"要求从给出的数据中寻找所关心的数据，它是对原始数据所包含信息的挖掘

### 1 实例描述 ###

给出child-parent（孩子——父母）表，要求输出grandchild-grandparent（孙子——祖父母）表  
样例输入：

child        parent
    Tom        Lucy
    Tom        Jack
    Jone        Lucy
    Jone        Jack
    Lucy        Mary
    Lucy        Ben
    Jack        Alice
    Jack        Jesse
    Terry        Alice
    Terry        Jesse
    Philip        Terry
    Philip        Alma
    Mark        Terry
    Mark        Alma

期望输出：

grandchild        grandparent 
    Tom            　　Alice
    Tom            　　Jesse
    Jone            　　Alice
    Jone           　　 Jesse
    Tom            　　Mary
    Tom            　　Ben
    Jone           　　 Mary
    Jone           　　 Ben
    Philip          　　  Alice
    Philip            　　Jesse
    Mark           　　 Alice
    Mark           　　 Jesse

### 2 问题分析 ###

这个实例需要进行单表连接，连接的是左表的parent列和右表的child列，且左表和右表是同一个表。  
　　连接结果中除去连接的两列就是所需要的结果——"grandchild--grandparent"表。要用MapReduce解决这个实例，首先应该考虑如何实现表的自连接；其次就是连接列的设置；最后是结果的整理  
MapReduce的shuffle过程会将相同的key会连接在一起，所以可以将map结果的key设置成待连接的列，然后列中相同的值就自然会连接在一起了。

### 3.实现步骤 ###

1.  map阶段将读入数据分割成child和parent之后，将parent设置成key，child设置成value进行输出，并作为左表；再将同一对child和parent中的child设置成key，parent设置成value进行输出，作为右表
2.  为了区分输出中的左右表，需要在输出的value中再加上左右表的信息，比如在value的String最开始处加上字符1表示左表，加上字符2表示右表
3.   reduce接收到连接的结果，其中每个key的value-list就包含了"grandchild--grandparent"关系。取出每个key的value-list进行解析，将左表中的child放入一个数组，右表中的parent放入一个数组，然后对两个数组求笛卡尔积就是最后的结果了

### 4.关键代码 ###

package com.mk.mapreduce;
    
    
    import org.apache.commons.lang.StringUtils;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.*;
    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.output.FileOutputFormat;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    import java.net.URI;
    import java.util.ArrayList;
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Objects;
    
    public class JoinSelf {
    
        public static class JoinSelfMapper extends Mapper<LongWritable, Text, Text, TableInfo> {
            @Override
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
                if (StringUtils.isBlank(value.toString())) {
                    System.out.println("空白行");
                    return;
                }
                String[] values = value.toString().split("\\s+");
                if (values.length < 2 || values[0].equals("child")) {
                    System.out.println("长度不够的行:" + value.toString());
                    return;
                }
    
                context.write(new Text(values[0]), new TableInfo(TableInfo.PARENT, values[1]));
                context.write(new Text(values[1]), new TableInfo(TableInfo.CHILD, values[0]));
            }
        }
    
        public static class JoinSelfReducer extends Reducer<Text, TableInfo, Text, Text> {
            @Override
            protected void setup(Context context) throws IOException, InterruptedException {
                context.write(new Text("grandchild"), new Text("grandparent"));
            }
    
            @Override
            protected void reduce(Text key, Iterable<TableInfo> values, Context context) throws IOException, InterruptedException {
    
                List<String> grandChildren = new LinkedList<>();
                List<String> grandParents = new LinkedList<>();
    
                for (TableInfo v : values) {
                    if (v.getTable() == TableInfo.CHILD) {
                        grandChildren.add(v.value.toString());
                    } else {
                        grandParents.add(v.value.toString());
                    }
    
                }
    
                if (!grandChildren.isEmpty() && !grandParents.isEmpty()) {
                    grandChildren.sort((a,b)->a.compareTo(b));
                    grandParents.sort((a,b)->a.compareTo(b));
                    for (String grandChild :grandChildren)
                        for (String grandParent : grandParents)
                            context.write(new Text(grandChild), new Text(grandParent));
    
                }
            }
        }
    
        public static class TableInfo implements WritableComparable<TableInfo> {
            public static final int PARENT = 1;
            public static final int CHILD = 2;
            private int table;
            private Text value;
    
            public TableInfo() {
            }
    
            public TableInfo(int table, String value) {
                this.table = table;
                this.value = new Text(value);
            }
    
            public int getTable() {
                return table;
            }
    
            public void setTable(int table) {
                this.table = table;
            }
    
            public void setValue(Text value) {
                this.value = value;
            }
    
            @Override
            public int compareTo(TableInfo o) {
                int c = table - o.table;
                if (c != 0)
                    return c;
                return value.compareTo(o.value);
            }
    
            @Override
            public void write(DataOutput out) throws IOException {
                out.writeInt(table);
                this.value.write(out);
            }
    
            @Override
            public void readFields(DataInput in) throws IOException {
                this.table = in.readInt();
                if (this.value == null)
                    this.value = new Text();
                this.value.readFields(in);
            }
    
            @Override
            public String toString() {
                return "TableInfo{" +
                        "table=\'" + table +
                        "\', value=\'" + value +
                        "\'}";
            }
        }
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    
            String uri = "hdfs://192.168.150.128:9000";
            String input = "/joinSelf/input";
            String output = "/joinSelf/output";
            Configuration conf = new Configuration();
            if (System.getProperty("os.name").toLowerCase().contains("win"))
                conf.set("mapreduce.app-submission.cross-platform", "true");
    
            FileSystem fileSystem = FileSystem.get(URI.create(uri), conf);
            Path path = new Path(output);
            fileSystem.delete(path, true);
    
            Job job = new Job(conf, "JoinSelf");
            job.setJar("./out/artifacts/hadoop_test_jar/hadoop-test.jar");
            job.setJarByClass(JoinSelf.class);
            job.setMapperClass(JoinSelfMapper.class);
            job.setReducerClass(JoinSelfReducer.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(TableInfo.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            FileInputFormat.addInputPaths(job, uri + input);
            FileOutputFormat.setOutputPath(job, new Path(uri + output));
    
    
            boolean ret = job.waitForCompletion(true);
            System.out.println(job.getJobName() + "-----" + ret);
        }
    }

编程中遇到的问题：

write String到hadoop的问题

read Text类初始化的问题

mapper空白字符\\u00A0分割字符串的问题

多个孙子与多祖父母的问题