mapreduce(四):单表关联查询

本文主要通过输入文件中的child字段和parent字段进行单表关联查询，推导出哪些用户具有child与grandparent关系。

1、 输入数据分析

输入文件数据示例：

child parent
Tom Jack 
Jack Alice
Jack Jesse
           

第1列表示child，第2列表示parent，我们需要根据child和parent的关系得出child和grantparent的关系。比如说Tom的parent是Jack，Jack的parent是Alice和Jesse，由此我们可以得出Tom的grantparent是{Alice，Jesse}。

2、 Map过程

首先使用默认的TextInputFormat类对输入文件进行处理，得到文本中每行的偏移量及其内容。Map过程首先将输入分割成child和parent，然后正序输出一次作为右表，反序输出一次作为左表，需要注意的是在输出的value中必须加上左右表区别标志，其中左表标识符为1，右表标识符为2，如图所示。

3、 Reduce过程

Reduce过程首先对输入< key,values >即<”Lucy”,[“1 Tom”,”2 Mary”,”2 Ben”]>的values值进行遍历获取到单元信息（例如”1 Tom”），然后将单元信息中的用户ID（例如Tom）按照左表、右表标识符分别存入到grandChild集合和grandParent集合，最后对grandChild集合和grandParent集合进行笛卡尔积运算得到child与grandParent的关系，并进行输出，如图所示。

<pre name="code" class="java">package com.mr5;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;

import com.mr1.EJob;
import com.mr3.CombinationKey;
import com.mr3.DefinedComparator;
import com.mr3.DefinedGroupSort;
import com.mr3.DefinedPartition;
import com.mr3.SecondSortMR;
import com.mr3.SecondSortMR.SortMapper;
import com.mr3.SecondSortMR.SortReducer;


/*
 * 输入文件数据示例：
child parent
Tom Jack 
Jack Alice
Jack Jesse
 * 
 * 输出结果：
 * Tom	   Alice
 * Tom	   Jesse
 * 
 * 
 * */


public class SingletonTableJoin {
	
	public static class JoinMapper extends Mapper<LongWritable, Text, Text, Text>{
		Text child=new Text();
		Text parent=new Text();
		String line="";
		@Override
		protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {
			if (value!=null) {
				line=value.toString();
				String[] values=line.split(" ");
				if (values.length>=2) {
					child.set(values[0]);
					parent.set("parent "+values[1]);
					context.write(child, parent);
					parent.set(values[1]);
					child.set("child "+values[0]);
					context.write(parent, child);
				}
			}
		}	
	}
	
	public static class JoinReducer extends Reducer<Text,Text,Text,Text>{
		
		String value="";
		HashSet<String> childs=new HashSet<>();
		HashSet<String> parents=new HashSet<>();
		Text child=new Text();
		Text parent=new Text();
		@Override
		protected void reduce(Text key, Iterable<Text> values,Context context)throws IOException, InterruptedException {
			for (Text text : values) {
				value=text.toString();
				if (value.contains("child")) {
					childs.add(value.split(" ")[1]);
				}
				if (value.contains("parent")) {
					parents.add(value.split(" ")[1]);
				}
			}
			
			if (childs.size()!=0&&parents.size()!=0) {
				for (String str1 : childs) {
					child.set(str1);
					for (String str2 : parents) {
						parent.set("   "+str2);
						context.write(child, parent);
					}
				}
			}
			childs.clear();
			parents.clear();
		}
		
	}
		public static void main(String[] args) throws Exception {
			File jarFile = EJob.createTempJar("bin");
			ClassLoader classLoader = EJob.getClassLoader();
			Thread.currentThread().setContextClassLoader(classLoader);
			
			Configuration conf = new Configuration(true);
			conf.set("fs.default.name", "hdfs://192.168.56.111:9000");
			conf.set("hadoop.job.user", "root");
			conf.set("mapreduce.framework.name", "yarn");
			conf.set("mapreduce.jobtracker.address", "192.168.56.111:9001");
			conf.set("yarn.resourcemanager.hostname", "192.168.56.111");
			conf.set("yarn.resourcemanager.admin.address", "192.168.56.111:8033");
			conf.set("yarn.resourcemanager.address", "192.168.56.111:8032");
			conf.set("yarn.resourcemanager.resource-tracker.address","192.168.56.111:8031");
			conf.set("yarn.resourcemanager.scheduler.address","192.168.56.111:8030");
			conf.setBoolean("mapreduce.app-submission.cross-platform", true);
			String[] otherArgs = new String[2];
			otherArgs[0] = "hdfs://192.168.56.111:9000/testdata/SingletonTableJoin.txt";// 计算原文件目录，需提前在里面存入文件
			String time = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());
			otherArgs[1] = "hdfs://192.168.56.111:9000/output/output" + time;// 计算后的计算结果存储目录，每次程序执行的结果目录不能相同，所以添加时间标签
			
			Job job = new Job(conf, "SingletonTableJoin job");
			job.setJarByClass(SingletonTableJoin.class);
			((JobConf) job.getConfiguration()).setJar(jarFile.toString());// 环境变量调用，添加此句则可在eclipse中直接提交mapreduce任务，如果将该java文件打成jar包，需要将该句注释掉，否则在执行时反而找不到环境变量
			System.out.println("Job start!");
			FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
			FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
			
			job.setMapperClass(JoinMapper.class);
			job.setReducerClass(JoinReducer.class);
			
			 //设置reduce的输出key和value类型 
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			
			
			if (job.waitForCompletion(true)) {
				System.out.println("ok!");
			} else {
				System.out.println("error!");
				System.exit(0);
			}
		}
		
	}