版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/u011410254/article/details/82903217
在hadoop入门3里,用订单和产品进行关联,用map+reducer实现join逻辑,但是这种使用,小数据下还好,但是一旦出现海量数据,会出现reduce处理任务严重不平衡,有的reduce很轻松,有的reduce很繁忙,也就是数据倾斜;因此去掉reduce这一步,直接在map完成join,
需要在map完成join过程,势必需要在每个map task里获取产品信息(产品信息是小部分,可以在放在每个mapTask里),产品放入每个map task,hadoop已经提供这种机制:
//制定缓存文件到所有的maptask运行节点
//job.addArchiveToClassPath(archive);//缓存jar包到task运行节点的calsspath中
//job.addFileToClassPath(file);//缓存普通文件到task运行节点的calsspath中
//job.addCacheArchive(uri);//缓存压缩包文件到task运行节点的工作目录
//job.addCacheFile(uri);//缓存普通文件到task运行节点的工作目录
//将产品信息缓存到task运行节点里
//job.addCacheFile(new URI("file:/e:/data/mapjoin/product/product.txt"));
job.addCacheFile(new URI("hdfs://hadoop01:9000/product/product.txt"));
完整代码:
package com.zsy.mr.mapjoin;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class MapSideJoin {
static class MapSideMapper extends Mapper<LongWritable, Text, Text, NullWritable>{
Map<String, String> productMap = new HashMap<String, String>();
Text v = new Text();
/**
* setup 是maptask处理数据之前调用,可以进行数据初始化
*/
@Override
protected void setup(Context context)
throws IOException, InterruptedException {
// String paths = context.getLocalCacheFiles()[0].getName();
BufferedReader bReader = new BufferedReader(new InputStreamReader(new FileInputStream("product.txt")));
List<String> list = IOUtils.readLines(bReader);
String[] tempStr = null;
for (String string : list) {
if(StringUtils.isNotBlank(string)) {
tempStr = string.split(" ");
productMap.put(tempStr[0].toString(), string);
}
}
}
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context)
throws IOException, InterruptedException {
//通过空格分割
String[] strs = value.toString().split(" ");
String pId = strs[2];//产品id
String resultProduct = productMap.get(pId);
v.set(value.toString()+" "+resultProduct);
context.write(v, NullWritable.get());
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(MapSideJoin.class);
job.setMapperClass(MapSideMapper.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
//FileInputFormat.setInputPaths(job, new Path("E:\\data\\mapjoin\\input"));
FileInputFormat.setInputPaths(job, new Path(args[0]));
//FileOutputFormat.setOutputPath(job, new Path("E:\\data\\mapjoin\\output"));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//制定缓存文件到所有的maptask运行节点
//job.addArchiveToClassPath(archive);//缓存jar包到task运行节点的calsspath中
//job.addFileToClassPath(file);//缓存普通文件到task运行节点的calsspath中
//job.addCacheArchive(uri);//缓存压缩包文件到task运行节点的工作目录
//job.addCacheFile(uri);//缓存普通文件到task运行节点的工作目录
//将产品信息缓存到task运行节点里
//job.addCacheFile(new URI("file:/e:/data/mapjoin/product/product.txt"));
job.addCacheFile(new URI("hdfs://hadoop01:9000/product/product.txt"));
job.setNumReduceTasks(0 );
boolean res = job.waitForCompletion(true);
System.exit(res?0:1);
}
}
hadoop集群运行结果:
可以正常join。
但是我遇到的一个问题,我在eclipse运行,在setUp里死活找不到product.txt文件,放到虚拟机的集群里跑就可以,不知道是啥原因,后面有时间看看这个问题