Map Join 实现方式一
● 使用场景：一个大表（整张表内存放不下，但表中的key内存放得下），一个超大表
● 实现方式：分布式缓存
● 用法:
SemiJoin就是所谓的半连接，其实仔细一看就是reduce join的一个变种，就是在map端过滤掉一些数据，在网络中只传输参与连接的数据不参与连接的数据不必在网络中进行传输，从而减少了shuffle的网络传输量，使整体效率得到提高，其他思想和reduce join是一模一样的。说得更加接地气一点就是将小表中参与join的key单独抽出来通过DistributedCach分发到相关节点，然后将其取出放到内存中(可以放到HashSet中)，在map阶段扫描连接表，将join key不在内存HashSet中的记录过滤掉，让那些参与join的记录通过shuffle传输到reduce端进行join操作，其他的和reduce join都是一样的。
代码实现
package com.hadoop.reducejoin.test; import java.io.BufferedReader; import java.io.FileNotFoundException; import java.io.FileReader; import java.io.IOException; import java.io.InputStreamReader; import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; 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.input.FileSplit; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; import org.apache.hadoop.util.bloom.Key; /* * 一个大表，一个小表（也很大，内存中放不下） * map 阶段：Semi Join解决小表整个记录内存放不下的场景，那么就取出来一小部分关键字段放入内存，过滤大表 * 提前过滤，提前提取出小表中的连接字段放入内存中，在map阶段就仅留下大表中那些小表中存在的连接字段key * reduce 阶段：reduce side join */ public class SemiJoin { /** * 为来自不同表(文件)的key/value对打标签以区别不同来源的记录。 * 然后用连接字段作为key，其余部分和新加的标志作为value，最后进行输出。 */ public static class SemiJoinMapper extends Mapper<Object, Text, Text, Text> { // 定义Set集合保存小表中的key private Set<String> joinKeys = new HashSet<String>(); private Text joinKey = new Text(); private Text combineValue = new Text(); /** * 获取分布式缓存文件 */ protected void setup(Context context) throws IOException, InterruptedException { BufferedReader br; String infoAddr = null; // 返回缓存文件路径 Path[] cacheFilesPaths = context.getLocalCacheFiles(); for (Path path : cacheFilesPaths) { String pathStr = path.toString(); br = new BufferedReader(new FileReader(pathStr)); while (null != (infoAddr = br.readLine())) { // 按行读取并解析气象站数据 String[] records = StringUtils.split(infoAddr.toString(), "\t"); if (null != records)// key为stationID joinKeys.add(records[0]); } } } @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { String pathName = ((FileSplit) context.getInputSplit()).getPath() .toString(); // 如果数据来自于records，加一个records的标记 if (pathName.endsWith("records-semi.txt")) { String[] valueItems = StringUtils.split(value.toString(), "\t"); // 过滤掉脏数据 if (valueItems.length != 3) { return; } // 提前过滤，提前提取出小表中的连接字段，在map阶段就仅留下大表中那些小表中存在的连接字段key if (joinKeys.contains(valueItems[0])) { joinKey.set(valueItems[0]); combineValue.set("records-semi.txt" + valueItems[1] + "\t" + valueItems[2]); context.write(joinKey, combineValue); } } else if (pathName.endsWith("station.txt")) { // 如果数据来自于station，加一个station的标记 String[] valueItems = StringUtils.split(value.toString(), "\t"); // 过滤掉脏数据 if (valueItems.length != 2) { return; } joinKey.set(valueItems[0]); combineValue.set("station.txt" + valueItems[1]); context.write(joinKey, combineValue); } } } /* * reduce 端做笛卡尔积 */ public static class SemiJoinReducer extends Reducer<Text, Text, Text, Text> { private List<String> leftTable = new ArrayList<String>(); private List<String> rightTable = new ArrayList<String>(); private Text result = new Text(); @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { // 一定要清空数据 leftTable.clear(); rightTable.clear(); // 相同key的记录会分组到一起，我们需要把相同key下来自于不同表的数据分开，然后做笛卡尔积 for (Text value : values) { String val = value.toString(); System.out.println("value=" + val); if (val.startsWith("station.txt")) { leftTable.add(val.replaceFirst("station.txt", "")); } else if (val.startsWith("records-semi.txt")) { rightTable.add(val.replaceFirst("records-semi.txt", "")); } } // 笛卡尔积 for (String leftPart : leftTable) { for (String rightPart : rightTable) { result.set(leftPart + "\t" + rightPart); context.write(key, result); } } } } public static void main(String[] arg0) throws Exception { Configuration conf = new Configuration(); String[] args = { "hdfs://sparks:9000/middle/reduceJoin/station.txt", "hdfs://sparks:9000/middle/reduceJoin/station.txt", "hdfs://sparks:9000/middle/reduceJoin/records-semi.txt", "hdfs://sparks:9000/middle/reduceJoin/SemiJoin-out" }; String[] otherArgs = new GenericOptionsParser(conf, args) .getRemainingArgs(); if (otherArgs.length < 2) { System.err.println("Usage: semijoin <in> [<in>...] <out>"); System.exit(2); } //输出路径 Path mypath = new Path(otherArgs[otherArgs.length - 1]); FileSystem hdfs = mypath.getFileSystem(conf);// 创建输出路径 if (hdfs.isDirectory(mypath)) { hdfs.delete(mypath, true); } Job job = Job.getInstance(conf, "SemiJoin"); //添加缓存文件 job.addCacheFile(new Path(otherArgs[0]).toUri()); job.setJarByClass(SemiJoin.class); job.setMapperClass(SemiJoinMapper.class); job.setReducerClass(SemiJoinReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); //添加输入路径 for (int i = 1; i < otherArgs.length - 1; ++i) { FileInputFormat.addInputPath(job, new Path(otherArgs[i])); } //添加输出路径 FileOutputFormat.setOutputPath(job, new Path( otherArgs[otherArgs.length - 1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }