本文共 3358 字，大约阅读时间需要 11 分钟。
Map Join
① Map Join工作原理
Map Join是一种在Map阶段进行的关联操作，适用于处理小表或大表的场景。
Map Join的主要优势在于减少Reduce阶段的数据倾斜风险。在Map端通过提前缓存多张表的数据，完成join逻辑的处理，从而将复杂的关联操作完全转移到Map阶段。
具体实现方式是通过DistributedCache来缓存所需的数据文件。在Mapper的setup阶段，读取并存储到缓存集合中。在Map阶段，通过缓存中的数据快速获取所需信息，完成join操作。
代码示例：
job.addCacheFile(new URI("file://e:/cache/pd.txt")); ② Map Join 案例
需求分析
Map Join特别适用于关联表中包含小表的情况。通过在Map阶段完成join操作，能够显著提升性能并减少资源占用。
代码实现：
Mapper阶段
在Mapper阶段，我们首先读取并缓存小表的数据。然后，在Map阶段根据Map端的业务逻辑，快速查找所需的关联信息，并将结果输出。
代码示例：
package 第三章_MR框架原理.多种join应用; import org.apache.commons.lang.StringUtils; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.BufferedReader; import java.io.FileInputStream; import java.io.IOException; import java.io.InputStreamReader; import java.net.URI; import java.util.HashMap; public class DistributedCacheMapper extends Mapper { Text k = new Text(); HashMap pdMap = new HashMap<>(); @Override protected void setup(Context context) throws IOException, InterruptedException { URI[] cacheFiles = context.getCacheFiles(); String path = cacheFiles[0].getPath().toString(); BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path), "UTF-8")); while (StringUtils.isNotEmpty(line = reader.readLine())) { String[] fields = line.split("\t"); pdMap.put(fields[0], fields[1]); } IOUtils.closeStream(reader); } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] fields = line.split("\t"); String pid = fields[1]; String pname = (String) pdMap.get(pid); line += "\t" + pname; k.set(line); context.write(k, NullWritable.get()); } }
Driver阶段
在Driver阶段，我们配置Job并设置Map阶段完成join操作。通过设置NumReduceTasks为0，确保join操作完全在Map阶段完成。
代码示例：
package 第三章_MR框架原理.多种join应用; import java.net.URI; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class DistributedCacheDriver { public static void main(String[] args) throws IOException { Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); job.setJarByClass(DistributedCacheDriver.class); job.setMapperClass(DistributedCacheMapper.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); FileInputFormat.setInputPaths(job, new Path("G:\Projects\IdeaProject-C\MapReduce\src\main\java\第三章_MR框架原理\多种join应用\data\order.txt")); FileOutputFormat.setOutputPath(job, new Path("G:\Projects\IdeaProject-C\MapReduce\src\main\java\第三章_MR框架原理\多种join应用\cacheoutput")); job.addCacheFile(new URI("file:///G:/Projects/IdeaProject-C/MapReduce/src/main/java/第三章_MR框架原理/多种join应用/data/pd.txt")); job.setNumReduceTasks(0); boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } } 总结
通过Map Join技术，我们能够在Map阶段完成join操作，显著提升性能并减少数据倾斜风险。这种方法在处理小表或大表时都能发挥优势，是Hadoop MapReduce优化的重要手段。
转载地址：http://hueq.baihongyu.com/