版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/u010800708/article/details/86425190
mapjoin指的是在mapreduce的map阶段先加载一个文件缓存到内存当中,这个文件可能是从磁盘读取的或网络请求的都可以。
map(key,value,context)方法中读取的数据key和value,这两个数据和先前缓存到内存中的数据一起做处理后再context.write()到reduce阶段。
mapjoin相当于在map阶段写数据到reduce阶段前对数据做了处理。
比如有两个txt文件
pd.txt
01 mac
02 huawei
03 xiaomi
第一个列代表订单id ,第二列代表商品名称
order.txt
201801 01 1
201802 02 2
201803 03 3
201804 01 4
201805 02 5
201806 03 6
第一列代表时间戳,第二列代表订单id,第三列代表数量
如果希望在map阶段输出
时间戳 订单id 订单数量 商品名称
201801 01 1 mac
可以在map中这样定义
package com.tony.mapjoin;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/*
* 商品表加载到内存中,然后数据在map端输出前进行替换
*/
public class CacheMapper extends Mapper<LongWritable,Text,Text,NullWritable> {
//1.商品表加入到内存
HashMap<String,String> pdmap = new HashMap<>();
protected void setup(Context context) throws IOException {
//加载缓存文件
BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("pd.txt"),"utf-8"));
String line;
while(StringUtils.isNotEmpty(line = br.readLine())) {
//切分
String[] fields = line.split("\t");
//缓存
pdmap.put(fields[0], fields[1]);
}
br.close();
}
//2.map传输
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//获取数据
String line = value.toString();
//切割
String[] fields = line.split("\t");
//获取订单商品id
String pid = fields[1];
//根据id查找商品名称
String pdName = pdmap.get(pid);
//拼接数据
line = line +"\t" +pdName;
//输出
context.write(new Text(line), NullWritable.get());
}
}
在自定义driver驱动类中
public class CacheDriver {
public static void main(String[] args) throws Exception {
//1.获取job信息
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
//2.获取jar包
job.setJarByClass(CacheDriver.class);
//3.获取自定义的mapper与reducer类
job.setMapperClass(CacheMapper.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
FileInputFormat.setInputPaths(job, new Path("e://bigdata//table0112//in"));
FileOutputFormat.setOutputPath(job, new Path("e://bigdata//table0112//out"));
//加载缓存商品数据
job.addCacheFile(new URI("file:///e:/bigdata/table0112/inputfile/pd.txt"));
//设置一下reducetask的数量,因为没有设置reduce类
job.setNumReduceTasks(0);
//提交任务
boolean rs = job.waitForCompletion(true);
System.out.println(rs0:1);
}
}