需求
数据：
【主表】：存放在log.txt中
--------------------------------------------------------
手机号码 品牌类型 登录时间 在线时长
13512435454 1 2018-11-12 12：32：32 50
.......
-------------------------------------------------------- 【从表】：存放在type.txt中
--------------------------------------------------------
品牌类型（主键） 品牌名称
1 动感地带
2 xxxxx
.......
-------------------------------------------------------- 目标输出：
--------------------------------------------------------------------
手机号码 品牌类型 品牌名称 登录时间 在线时长
13512435454 1 动感地带 2018-11-12 12：32：32 50
.......
--------------------------------------------------------------------
测试数据
type.txt（type表）
1 动感地带
2 全球通
3 神州行
4 神州大众卡
7 流量王
log.txt（log表）
13112345123 1 2018-11-11 00:00:00 50
13245612378 1 2018-11-11 12:32:45 18
13674589656 5 2018-11-12 13:25:15 66
13192258656 2 2018-11-14 07:05:15 12
13747958635 4 2018-11-15 09:12:59 47
13565412545 3 2018-11-16 13:04:09 19
注：数据均以TAB键划分
目标输出
13245612378 1 动感地带 2018-11-11 12:32:45 18
13112345123 1 动感地带 2018-11-11 00:00:00 50
13192258656 2 全球通 2018-11-14 07:05:15 12
13565412545 3 神州行 2018-11-16 13:04:09 19
13747958635 4 神州大众卡 2018-11-15 09:12:59 47
13674589656 5 null 2018-11-12 13:25:15 66
实现方式一：Reducer端的join实现
思路
- 在Mapper阶段：将 type.txt 和 log.txt 放在同一个文件夹上，通过判断输入文件的路径来判断数据来自哪个表
- 对于type表的数据就输出<品牌类型，“t”+品牌名称>
- 对于log表的数据就输出<品牌类型, "l"+手机号码+“\t’”+登录时间+“\t’”+在线时长>
- 在Reducer阶段：由于Mapper输出的Key为品牌类型，那么两个表中同一品牌类型的数据就会在一次reduce函数被调用时被处理，同时由于品牌类型是type表的主键，所以reduce函数处理的数据中至多有一个value来自type表，因此可以遍历整个value-list，将对应Key的品牌名称以及对应log表的数据保存起来，然后再遍历收集到的来自log表的数据将Key值对应的品牌名称数据插入到每一行中即可。
代码实现
package test.linzch3; import java.io.IOException;
import java.util.LinkedList; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
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 org.apache.hadoop.util.GenericOptionsParser; public class LeftOuterJoin1 {
private static class MyMapper extends Mapper<Object, Text, Text, Text>{ private final Text outKey = new Text();
private final Text outVal = new Text(); @Override
public void map(Object key, Text value, Context context) throws IOException, InterruptedException{
String line = value.toString();
if(line == null || line.trim().equals("")) return;//抛弃空记录
String filePath = ((FileSplit) context.getInputSplit()).getPath().toString();
String[] values = line.split("\t");
//根据输入文件路径分别处理type.txt和log.txt，用"t"和"l"标记两个表的value
if(filePath.contains("type.txt") && values.length == 2){
outKey.set(values[0]);
outVal.set("t" + values[1]);
context.write(outKey, outVal);
}else if(filePath.contains("log.txt") && values.length == 4){
outKey.set(values[1]);
outVal.set("l" + values[0] + "\t" + values[2] + "\t" + values[3]);
context.write(outKey, outVal);
}
}
} private static class MyReducer extends Reducer<Text, Text, Text, Text> { private LinkedList<String> logs = new LinkedList<String>();
private String type = "";
private final Text outKey = new Text();
private final Text outVal = new Text(); @Override
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
logs.clear();
type = "Null"; //默认为Null //根据value的第一个标记字符判断是type表的数据还是log表的数据
for(Text tval:values){
String val = tval.toString();
if(val.startsWith("l"))
logs.add(val.substring(1));
else if(val.startsWith("t"))
type = val.substring(1);
} for(String log:logs){
String[] fields = log.split("\t");
outKey.set(fields[0]);
outVal.set(key.toString() + "\t" + type + "\t" + fields[1] + "\t" + fields[2]);
context.write(outKey, outVal);
}
}
} public static void main(String[] args) throws Exception{
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: LeftOuterJoin <in> <out>");
System.exit(2);
}
Job job = Job.getInstance(conf, "Left outer join1");
job.setJarByClass(LeftOuterJoin1.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class); job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); /*delete the output directory if exists*/
Path out = new Path(otherArgs[otherArgs.length - 1]);
FileSystem fileSystem = FileSystem.get(conf);
if (fileSystem.exists(out)) {
fileSystem.delete(out, true);
} System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
实现方式二：Mapper端的join实现
思路
- 当join的两个表中有一个表数据量不大，可以轻松加载到各节点内存中时，可以使用DistributedCache将小表的数据加载到分布式缓存，然后MapReduce框架会缓存数据分发到需要执行map任务的节点上，在map节点上直接调用本地的缓存文件参与计算。在Map端完成join操作，可以降低网络传输到Reduce端的数据流量，有利于提高整个作业的执行效率。
- 假设type表数据量较小，则将type.txt的数据添加到DistributedCache中，在map计算中读取本地缓存的type.txt数据并将对应log表中的每一行数据插入对应品牌类型的品牌名称，这里无需实现Reducer。
代码实现
package test.linzch3; import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.LinkedList;
import java.util.Map; import org.apache.commons.collections.map.HashedMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
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 org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.yarn.api.records.URL; public class LeftOuterJoin2 {
private static class MyMapper extends Mapper<Object, Text, Text, Text>{ private Map<String, String> typeMaps = new HashedMap();
private final Text outKey = new Text();
private final Text outVal = new Text(); @Override
protected void setup(Context context) throws IOException ,InterruptedException {
//此处使用快捷方式type.txt访问
FileReader fr = new FileReader("type.txt");
BufferedReader br = new BufferedReader(fr);
String line;
while((line = br.readLine()) != null) {
//map端加载缓存数据
String[] values = line.split("\t");
if(values.length != 2) continue;
typeMaps.put(values[0], values[1]);
}
}; @Override
public void map(Object key, Text value, Context context) throws IOException, InterruptedException{
String line = value.toString();
if(line == null || line.trim().equals("")) return;//抛弃空记录
String[] values = line.split("\t");
outKey.set(values[0]);
outVal.set(values[1] + "\t" + typeMaps.get(values[1]) + "\t" + values[2] + "\t" + values[3]);
context.write(outKey, outVal);
}
} private final static String FILE_IN_PATH = "hdfs://localhost:9000/user/hadoop/input2/log.txt";
private final static String FILE_OUT_PATH = "hdfs://localhost:9000/user/hadoop/output2/"; public static void main(String[] args) throws Exception{
Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "Left outer join2"); job.addCacheFile(new URI("hdfs://localhost:9000/user/hadoop/input2/type.txt"));//添加分布式缓存文件 可以在map或reduce中直接通过type.txt链接访问对应缓存文件
job.setJarByClass(LeftOuterJoin1.class);
job.setMapperClass(MyMapper.class); job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(FILE_IN_PATH));
FileOutputFormat.setOutputPath(job, new Path(FILE_OUT_PATH)); /*delete the output directory if exists*/
Path out = new Path(FILE_OUT_PATH);
FileSystem fileSystem = FileSystem.get(conf);
if (fileSystem.exists(out)) {
fileSystem.delete(out, true);
} System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
实现方式三：二次排序版实现
思路
- 思考上面的两种实现方式
- 实现方式一：在Reducer端，假设输入的value-list很长很长，按照这种方式实现，遍历整个value-list找到对应Key（品牌类型）的品牌名称这单个数据（暂且称为数据A）并且将属于log表的数据都暂时保存到一个LinkedList上，时间和存储上的开销都随着value-list的长度增加而增长，这显然不适合大量数据的场合
- 实现方式二：虽然在Mapper端可以直接读文件，这样的处理确实是比较高效的，但是其前提是type表可以分布式缓存到各个节点上，但是一旦两个表都很大无法缓存到所有节点，这样该方式就失效了
- 总结：对于实现方式二，文件一大就不行了，这是无法优化的。但是对于实现方式一，其实还有优化的余地，实现方式一的问题就在于要遍历整个value-list的开销很大，而之所以要遍历整个value-list的原因便是为了数据A，那么有没有办法不用遍历就可以找到整个值呢？答案就是利用二次排序。
- 优化思路
- 要想不用遍历就可以找到数据A，那么问题就等价于 在这个value-list中，我们事先就知道数据A在value-list的位置了，很明显的两个位置就是：value-list的第一个和最后一个，而如果是第一个的话，那么我们在reduce函数每次都只用判断第一个value是否来自type表，剩下的就迭代value-list输出即可，这样甚至都不用保存log表数据，时间和存储上都一并优化了。
- 那么，如何让数据A能保持在value-list的第一个呢？这里就要利用MR的magic field——shuffle阶段了，具体操作如下：
- 设计组合Key：<数据类型tag, 品牌类型brandType>，两者都是Int型数据，tag的数据只有0或者1（type表的数据对应0，log表的数据对应1）
- 自定义实现分区类和分组类：让属于同个brandType的数据（不管来自哪个表）都能在同一个Reducer的一次函数调用被一并处理
- Mapper端：和实现方式一的Mapper的原理一样
- Reducer端：先判断value-list的第一个数据是否来自type表（若没有，数据A就默认是null），然后再遍历value-list输出剩余log表的所有数据（插入数据A在每一行中）
代码实现
package test.linzch3; import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Iterator; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
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; public class LeftOuterJoin3 { public static class CompositeKey implements WritableComparable<CompositeKey>{ private int tag;
private int brandType; public int getTag() {
return tag;
} public int getBrandType() {
return brandType;
} public void set(int tag_, int brandType_){
tag = tag_;
brandType = brandType_;
} @Override
public void write(DataOutput out) throws IOException { out.writeInt(tag);
out.writeInt(brandType);
} @Override
public void readFields(DataInput in) throws IOException { tag = in.readInt();
brandType = in.readInt();
} @Override
public int compareTo(CompositeKey other) { if(brandType != other.brandType)
return brandType < other.brandType ? -1 : 1;
else if(tag != other.tag)
return tag < other.tag ? -1: 1;
else return 0;
} } private static class MyPartitioner extends Partitioner<CompositeKey, IntWritable>{ @Override
public int getPartition(CompositeKey key, IntWritable value,
int numPartitions) { return key.getBrandType() % numPartitions;
} } private static class MyGroupingComparator extends WritableComparator{ protected MyGroupingComparator()
{
super(CompositeKey.class, true);
} @Override
public int compare(WritableComparable w1, WritableComparable w2)
{
CompositeKey key1 = (CompositeKey) w1;
CompositeKey key2 = (CompositeKey) w2;
int l = key1.getBrandType();
int r = key2.getBrandType();
return l == r ? 0 : (l < r ? -1 : 1);
} }
private static class MyMapper extends Mapper<Object, Text, CompositeKey, Text>{ private final CompositeKey outKey = new CompositeKey();
private final Text outVal = new Text(); @Override
protected void map(Object key, Text value, Context context)
throws IOException, InterruptedException { String line = value.toString();
if(line == null || line.trim().equals("")) return;//抛弃空记录
String filePath = ((FileSplit) context.getInputSplit()).getPath().toString();
String[] values = line.split("\t");
//根据输入文件路径分别处理type.txt和log.txt，用"t"和"l"标记两个表的value
if(filePath.contains("type.txt") && values.length == 2){
outKey.set(0, Integer.valueOf(values[0]));
outVal.set(values[1]);
context.write(outKey, outVal);
}else if(filePath.contains("log.txt") && values.length == 4){
outKey.set(1, Integer.valueOf(values[1]));
outVal.set(values[0] + "\t" + values[2] + "\t" + values[3]);
context.write(outKey, outVal);
}
}
} private static class MyReducer extends Reducer<CompositeKey, Text, Text, Text> { private String type = "";
private final Text outKey = new Text();
private final Text outVal = new Text(); @Override
protected void reduce(CompositeKey key, Iterable<Text> values, Context context)
throws IOException, InterruptedException { Iterator<Text> it = values.iterator();
String val = it.next().toString(); //根据第一个value对应的key判断第一个数据是否来自type表
if(key.getTag() == 0){
type = val;
}else{
type = "null";
String[] fields = val.split("\t");
outKey.set(fields[0]);
outVal.set(key.brandType + "\t" + type + "\t" + fields[1] + "\t" + fields[2]);
context.write(outKey, outVal);
} while(it.hasNext()){
val = it.next().toString();
String[] fields = val.split("\t");
outKey.set(fields[0]);
outVal.set(key.brandType + "\t" + type + "\t" + fields[1] + "\t" + fields[2]);
context.write(outKey, outVal);
}
}
} public static void main(String[] args) throws Exception{
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: LeftOuterJoin <in> <out>");
System.exit(2);
}
Job job = Job.getInstance(conf, "Left outer join3");
job.setJarByClass(LeftOuterJoin3.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class); job.setPartitionerClass(MyPartitioner.class);
job.setGroupingComparatorClass(MyGroupingComparator.class); job.setMapOutputKeyClass(CompositeKey.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); /*delete the output directory if exists*/
Path out = new Path(otherArgs[otherArgs.length - 1]);
FileSystem fileSystem = FileSystem.get(conf);
if (fileSystem.exists(out)) {
fileSystem.delete(out, true);
} System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}