MapReduce表连接操作之Map端join

心已赠人 2022-06-07 03:06 212阅读 0赞

**一：背景**

MapReduce提供了表连接操作其中包括**Map端join**、**Reduce端join**还有**半连接**，现在我们要讨论的是Map端join，Map端join是指数据到达map处理函数之前进行合并的，效率要远远高于Reduce端join，因为Reduce端join是把所有的数据都经过Shuffle，非常消耗资源。

**二：技术实现**

**基本思路：**

**(1)：**需要join的两个文件，一个存储在HDFS中，一个使用DistributedCache.addCacheFile()将需要join的另外一个文件加入到所有Map缓存中。

**(2)：**在Map函数里读取该文件，进行join

**(3)：**将结果输出到reduce

**(4)：**DistributedCache.addCacheFile()需要在作业提交前设置。

**什么是DistributedCache？**

DistributedCache是为了方便用户进行应用程序开发而设计的文件分发工具。它能够将只读的外部文件进行自动分发到各个节点上进行本地缓存，以便task运行时加载。

**DistributedCache的使用步骤**

**(1)：**在HDFS中上传文件(文本文件、压缩文件、jar包等)

**(2)：**调用相关API添加文件信息

**(3)：**task运行前直接调用文件读写API获取文件。

**常见API：**

DistributedCache.addCacheFile();

DistributedCache.addCacheArchive();

下面我们通过一个示例来深入体会Map端join。

**表一：tb\_a数据如下**

**\[java\]**  [view plain][]   [copy][view plain]

1.  name    sex age depNo  
2.  zhang   male    20  1  
3.  li  female  25  2  
4.  wang    female  30  3  
5.  zhou    male    35  2

表二：tb\_b数据如下

**\[java\]**  [view plain][]   [copy][view plain]

1.  depNo   depName  
2.  1   sales  
3.  2   Dev  
4.  3   Mgt

\#需求就是连接上面两张表

**注意：在Map端join操作中，我们往往将较小的表添加到内存中，因为内存的资源是很宝贵的，这也说明了另外一个问题，那就是如果表的数据量都非常大则不适合使用Map端join。**

代码如下：

**\[java\]**  [view plain][]   [copy][view plain]

1.  **public** **class** MyMapJoin \{  
2.      // 定义输入路径  
3.      **private** **static** String INPUT\_PATH1 = "";  
4.      //加载到内存的表的路径  
5.      **private** **static** String INPUT\_PATH2 = "";  
6.      // 定义输出路径  
7.      **private** **static** String OUT\_PATH = "";  
8.    
9.      **public** **static** **void** main(String\[\] args) \{  
10.   
11.         **try** \{  
12.             // 创建配置信息  
13.             Configuration conf = **new** Configuration();  
14.             // 获取命令行的参数  
15.             String\[\] otherArgs = **new** GenericOptionsParser(conf, args).getRemainingArgs();  
16.             // 当参数违法时，中断程序  
17.             **if** (otherArgs.length != 3) \{  
18.                 System.err.println("Usage:MyMapJoin<in1> <in2> <out>");  
19.                 System.exit(1);  
20.             \}  
21.   
22.             // 给路径赋值  
23.             INPUT\_PATH1 = otherArgs\[0\];  
24.             INPUT\_PATH2 = otherArgs\[1\];  
25.             OUT\_PATH = otherArgs\[2\];  
26.             // 创建文件系统  
27.             FileSystem fileSystem = FileSystem.get(**new** URI(OUT\_PATH), conf);  
28.             // 如果输出目录存在，我们就删除  
29.             **if** (fileSystem.exists(**new** Path(OUT\_PATH))) \{  
30.                 fileSystem.delete(**new** Path(OUT\_PATH), **true**);  
31.             \}  
32.             // 添加到内存中的文件(随便添加多少个文件)  
33.             DistributedCache.addCacheFile(**new** Path(INPUT\_PATH2).toUri(), conf);  
34.   
35.             // 创建任务  
36.             Job job = **new** Job(conf, MyMapJoin.**class**.getName());  
37.             // 打成jar包运行，这句话是关键  
38.             job.setJarByClass(MyMapJoin.**class**);  
39.             //1.1 设置输入目录和设置输入数据格式化的类  
40.             FileInputFormat.setInputPaths(job, INPUT\_PATH1);  
41.             job.setInputFormatClass(TextInputFormat.**class**);  
42.   
43.             //1.2 设置自定义Mapper类和设置map函数输出数据的key和value的类型  
44.             job.setMapperClass(MapJoinMapper.**class**);  
45.             job.setMapOutputKeyClass(NullWritable.**class**);  
46.             job.setMapOutputValueClass(Emp\_Dep.**class**);  
47.   
48.             //1.3 设置分区和reduce数量  
49.             job.setPartitionerClass(HashPartitioner.**class**);  
50.             job.setNumReduceTasks(0);  
51.   
52.             FileOutputFormat.setOutputPath(job, **new** Path(OUT\_PATH));  
53.             // 提交作业 退出  
54.             System.exit(job.waitForCompletion(**true**) ? 0 : 1);  
55.   
56.         \} **catch** (Exception e) \{  
57.             e.printStackTrace();  
58.         \}  
59.     \}  
60.   
61.     **public** **static** **class** MapJoinMapper **extends** Mapper<LongWritable, Text, NullWritable, Emp\_Dep> \{  
62.   
63.         **private** Map<Integer, String> joinData = **new** HashMap<Integer, String>();  
64.   
65.         @Override  
66.         **protected** **void** setup(Mapper<LongWritable, Text, NullWritable, Emp\_Dep>.Context context) **throws** IOException, InterruptedException \{  
67.             // 预处理把要关联的文件加载到缓存中  
68.             Path\[\] paths = DistributedCache.getLocalCacheFiles(context.getConfiguration());  
69.             // 我们这里只缓存了一个文件，所以取第一个即可，创建BufferReader去读取  
70.             BufferedReader reader = **new** BufferedReader(**new** FileReader(paths\[0\].toString()));  
71.   
72.             String str = **null**;  
73.             **try** \{  
74.                 // 一行一行读取  
75.                 **while** ((str = reader.readLine()) != **null**) \{  
76.                     // 对缓存中的表进行分割  
77.                     String\[\] splits = str.split("\\t");  
78.                     // 把字符数组中有用的数据存在一个Map中  
79.                     joinData.put(Integer.parseInt(splits\[0\]), splits\[1\]);  
80.                 \}  
81.             \} **catch** (Exception e) \{  
82.                 e.printStackTrace();  
83.             \} **finally**\{  
84.                 reader.close();  
85.             \}  
86.   
87.         \}  
88.   
89.         @Override  
90.         **protected** **void** map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, Emp\_Dep>.Context context) **throws** IOException,  
91.                 InterruptedException \{  
92.             // 获取从HDFS中加载的表  
93.             String\[\] values = value.toString().split("\\t");  
94.             // 创建Emp\_Dep对象  
95.             Emp\_Dep emp\_Dep = **new** Emp\_Dep();  
96.             // 设置属性  
97.             emp\_Dep.setName(values\[0\]);  
98.             emp\_Dep.setSex(values\[1\]);  
99.             emp\_Dep.setAge(Integer.parseInt(values\[2\]));  
100.             // 获取关联字段depNo，这个字段是关键  
101.             **int** depNo = Integer.parseInt(values\[3\]);  
102.             // 根据depNo从内存中的关联表中获取要关联的属性depName  
103.             String depName = joinData.get(depNo);  
104.             // 设置depNo  
105.             emp\_Dep.setDepNo(depNo);  
106.             // 设置depName  
107.             emp\_Dep.setDepName(depName);  
108.   
109.             // 写出去  
110.             context.write(NullWritable.get(), emp\_Dep);  
111.         \}  
112.     \}  
113. \}

程序运行的结果：

![20150121172010250][]

[view plain]: http://blog.csdn.net/lzm1340458776/article/details/42971075#
[20150121172010250]: /images/20220607/cf7d8aa8fb7444c685ea791df3c4766c.png