MapReduce编程(五) 单表关联

简介: 一、问题描述下面给出一个child-parent的表格,要求挖掘其中的父子辈关系,给出祖孙辈关系的表格。输入文件内容如下:child parentSteven LucySteven Jack...

一、问题描述

下面给出一个child-parent的表格,要求挖掘其中的父子辈关系,给出祖孙辈关系的表格。

输入文件内容如下:

child    parent
Steven   Lucy
Steven   Jack
Jone     Lucy
Jone     Jack
Lucy     Mary
Lucy     Frank
Jack     Alice
Jack     Jesse
David    Alice
David    Jesse
Philip   David
Philip   Alma
Mark     David
Mark     Alma

根据父辈和子辈挖掘爷孙关系。比如:

Steven   Jack
Jack     Alice
Jack     Jesse

根据这三条记录,可以得出Jack是Steven的长辈,而Alice和Jesse是Jack的长辈,很显然Steven是Alice和Jesse的孙子。挖掘出的结果如下:

grandson    grandparent
Steven      Jesse
Steven      Alice

要求通过MapReduce挖掘出所有的爷孙关系。

二、分析

解决这个问题要用到一个小技巧,就是单表关联。具体实现步骤如下,Map阶段每一行的key-value输入,同时也把value-key输入。以其中的两行为例:

Steven   Jack
Jack     Alice

key-value和value-key都输入,变成4行:

Steven   Jack
Jack     Alice
Jack     Steven  
Alice    Jack

shuffle以后,Jack作为key值,起到承上启下的桥梁作用,Jack对应的values包含Alice、Steven,这时候Alice和Steven肯定是爷孙关系。为了标记哪些是孙子辈,哪些是爷爷辈,可以在Map阶段加上前缀,比如小辈加上前缀”-“,长辈加上前缀”+”。加上前缀以后,在Reduce阶段就可以根据前缀进行分类。

三、MapReduce程序

package com.javacore.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.StringTokenizer;


/**
 * Created by bee on 3/29/17.
 */
public class RelationShip {

    public static class RsMapper extends Mapper<Object, Text, Text, Text> {

        private static int linenum = 0;

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (linenum == 0) {
                ++linenum;
            } else {
                StringTokenizer tokenizer = new StringTokenizer(line, "\n");
                while (tokenizer.hasMoreElements()) {
                    StringTokenizer lineTokenizer = new StringTokenizer(tokenizer.nextToken());
                    String son = lineTokenizer.nextToken();
                    String parent = lineTokenizer.nextToken();
                    context.write(new Text(parent), new Text(
                            "-" + son));
                    context.write(new Text(son), new Text
                            ("+" + parent));
                }
            }

        }
    }

    public static class RsReducer extends Reducer<Text, Text, Text, Text> {
        private static int linenum = 0;

        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            if (linenum == 0) {
                context.write(new Text("grandson"), new Text("grandparent"));
                ++linenum;
            }
            ArrayList<Text> grandChild = new ArrayList<Text>();
            ArrayList<Text> grandParent = new ArrayList<Text>();

            for (Text val : values) {
                String s = val.toString();

                if (s.startsWith("-")) {
                    grandChild.add(new Text(s.substring(1)));
                } else {
                    grandParent.add(new Text(s.substring(1)));
                }
            }

            for (Text text1 : grandChild) {
                for (Text text2 : grandParent) {
                    context.write(text1, text2);
                }
            }


        }


    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        FileUtil.deleteDir("output");
        Configuration cong = new Configuration();

        String[] otherArgs = new String[]{"input/relations/table.txt",
                "output"};
        if (otherArgs.length != 2) {
            System.out.println("参数错误");
            System.exit(2);
        }

        Job job = Job.getInstance();
        job.setJarByClass(RelationShip.class);
        job.setMapperClass(RsMapper.class);
        job.setReducerClass(RsReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }
}

四、输出结果

grandson    grandparent
Mark    Jesse
Mark    Alice
Philip  Jesse
Philip  Alice
Jone    Jesse
Jone    Alice
Steven  Jesse
Steven  Alice
Steven  Frank
Steven  Mary
Jone    Frank
Jone    Mary
目录
相关文章
|
7月前
|
分布式计算 Hadoop Java
MapReduce编程:自定义分区和自定义计数器
MapReduce编程:自定义分区和自定义计数器
105 0
|
7月前
|
存储 分布式计算 负载均衡
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
【大数据技术Hadoop+Spark】MapReduce概要、思想、编程模型组件、工作原理详解(超详细)
238 0
|
7月前
|
存储 分布式计算 算法
【底层服务/编程功底系列】「大数据算法体系」带你深入分析MapReduce算法 — Shuffle的执行过程
【底层服务/编程功底系列】「大数据算法体系」带你深入分析MapReduce算法 — Shuffle的执行过程
107 0
|
4月前
|
分布式计算 大数据 Hadoop
揭秘MapReduce背后的魔法:从基础类型到高级格式,带你深入理解这一大数据处理利器的奥秘与实战技巧,让你从此不再是编程门外汉!
【8月更文挑战第17天】MapReduce作为分布式计算模型,是大数据处理的基石。它通过Map和Reduce函数处理大规模数据集,简化编程模型,使开发者聚焦业务逻辑。MapReduce分单阶段和多阶段,支持多种输入输出格式如`TextInputFormat`和`SequenceFileInputFormat`。例如,简单的单词计数程序利用`TextInputFormat`读取文本行并计数;而`SequenceFileInputFormat`适用于高效处理二进制序列文件。合理选择类型和格式可有效解决大数据问题。
74 1
|
6月前
|
分布式计算 Hadoop Java
MapReduce编程模型——在idea里面邂逅CDH MapReduce
MapReduce编程模型——在idea里面邂逅CDH MapReduce
90 15
|
6月前
|
分布式计算 Hadoop Java
Hadoop MapReduce编程
该教程指导编写Hadoop MapReduce程序处理天气数据。任务包括计算每个城市ID的最高、最低气温、气温出现次数和平均气温。在读取数据时需忽略表头,且数据应为整数。教程中提供了环境变量设置、Java编译、jar包创建及MapReduce执行的步骤说明,但假设读者已具备基础操作技能。此外,还提到一个扩展练习,通过分区功能将具有相同尾数的数字分组到不同文件。
66 1
|
6月前
|
存储 分布式计算 Hadoop
MapReduce编程模型——自定义序列化类实现多指标统计
MapReduce编程模型——自定义序列化类实现多指标统计
52 0
|
7月前
|
分布式计算 资源调度 Hadoop
MapReduce分布式编程
MapReduce分布式编程
84 1
|
6月前
|
机器学习/深度学习 分布式计算 并行计算
MapReduce是一种用于并行计算的编程模型和处理大规模数据集的实现
MapReduce是一种用于并行计算的编程模型和处理大规模数据集的实现
95 0
|
7月前
|
分布式计算 并行计算 Java
【分布式计算框架】 MapReduce编程初级实践
【分布式计算框架】 MapReduce编程初级实践
228 2
下一篇
DataWorks