hadoop入门4：Map实现Join逻辑，无需要使用reducer

本文主要是介绍hadoop入门4：Map实现Join逻辑，无需要使用reducer，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在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处理数据之前调用，可以进行数据初始化*/@Overrideprotected 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);}}}@Overrideprotected 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];//产品idString 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集群运行结果：