MapReduce应用案例--单表关联

1. 实例描述

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

　　实例中给出child-parent 表， 求出grandchild-grandparent表。

　　输入数据 file01:

child        parent

Tom          Lucy

Tom          Jack

Jone         Lucy

Jone         Jack

Lucy         Marry

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    Marry

Tom    Ben

Jone    Marry

Jone    Ben

Philip    Alice

Philip    Jesse

Mark    Alice

Mark    Jesse

2. 设计思路

　　1. 在map阶段，将原数据进行分割，将parent作为map输出的key值，child作为map输出的value值，这样形成左表。

　　2. 同时在map阶段过程中，将child作为map输出的key值，parent作为map输出的value值，这样形成右表。

　　3. 连接左表的paren列和右表的child列。

3. 具体实现

　　

package tablerelation;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

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;

/**

 *

 * @author Amei 单表链接，求grandchild grandparent表

 */

public class SingleTableRelation {

    public static int time = 0;

    /**

     *

     * @author Amei 左表的paren 和 右表的 child 做链接

     */

    public static class Map extends Mapper<LongWritable, Text, Text, Text> {

        protected void map(LongWritable key, Text value, Context context)

                throws java.io.IOException, InterruptedException {
　　　　　　　// 左右表的标识

            int relation;

            StringTokenizer tokenizer = new StringTokenizer(value.toString());

            String child = tokenizer.nextToken();

            String parent = tokenizer.nextToken();

            if (child.compareTo("child") != 0) {

                // 左表

                relation = 1;

                context.write(new Text(parent),

                        new Text(relation + "+" + child));

                // 右表

                relation = 2;

                context.write(new Text(child),

                        new Text(relation + "+" + parent));

            }

        };

    }

    public static class Reduce extends Reducer<Text, Text, Text, Text> {

        protected void reduce(Text key, Iterable<Text> values,

                Reducer<Text, Text, Text, Text>.Context output)

                throws java.io.IOException, InterruptedException {

            int grandchildnum = 0;

            int grandparentnum = 0;

            List<String> grandchilds = new ArrayList<>();

            List<String> grandparents = new ArrayList<>();

            /** 输出表头 */

            if (time == 0) {

                output.write(new Text("grandchild"), new Text("grandparent"));

                time++;

            }

            for (Text val : values) {

                String record = val.toString();

                char relation = record.charAt(0);

                // 取出此时key所对应的child

                if (relation == '1') {

                    String child = record.substring(2);

                    grandchilds.add(child);

                    grandchildnum++;

                }

                // 取出此时key所对应的parent

                else {

                    String parent = record.substring(2);

                    grandparents.add(parent);

                    grandparentnum++;

                }

            }

            if (grandchildnum != 0 && grandparentnum != 0) {

                for (int i = 0; i < grandchildnum; i++)

                    for (int j = 0; j < grandparentnum; j++)

                        output.write(new Text(grandchilds.get(i)), new Text(

                                grandparents.get(j)));

            }

        }

    }

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

        Configuration conf = new Configuration();

        Job job =  new Job(conf,"single tale relation");

        job.setJarByClass(SingleTableRelation.class);

        job.setMapperClass(Map.class);

        job.setReducerClass(Reduce.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path("/user/hadoop_admin/singletalein"));

        FileOutputFormat.setOutputPath(job, new Path("/user/hadoop_admin/singletableout"));

        System.exit((job.waitForCompletion(true) ? 0 : 1));

    }

}