MapReduce实现等值连接，左外连接，右外连接，全外连接

Posted 2020-06-12

#测试数据：

# more user.txt(用户ID，用户名)

 

[java] view plain copy

 

1. 1   lavimer  
2. 2   liaozhongmin  
3. 3   liaozemin  

#more post.txt(用户ID，帖子ID，标题)

 

 

[java] view plain copy

 

1. 1   1   java  
2. 1   2   c  
3. 2   3   hadoop  
4. 4   4   hive  
5. 5   5   hbase  
6. 5   6   pig  
7. 5   7   flume  

#等值连接结果如下：

 

 

[java] view plain copy

 

1. 1   lavimer 1   1   java  
2. 1   lavimer 1   2   c  
3. 2   liaozhongmin    2   3   hadoop  

#左外连接结果如下：

 

 

[java] view plain copy

 

1. 1   lavimer 1   1   java  
2. 1   lavimer 1   2   c  
3. 2   liaozhongmin    2   3   hadoop  
4. 3   liaozemin   NULL  

#右外连接结果如下：

 

 

[java] view plain copy

 

1. 1   lavimer 1   1   java  
2. 1   lavimer 1   2   c  
3. 2   liaozhongmin    2   3   hadoop  
4. NULL    4   4   hive  
5. NULL    5   5   hbase  
6. NULL    5   6   pig  
7. NULL    5   7   flume  

#全外连接结果如下：

 

 

[java] view plain copy

 

1. 1   lavimer 1   1   java  
2. 1   lavimer 1   2   c  
3. 2   liaozhongmin    2   3   hadoop  
4. 3   liaozemin   NULL  
5. NULL    4   4   hive  
6. NULL    5   5   hbase  
7. NULL    5   6   pig  
8. NULL    5   7   flume  

实现代码如下：

 

 

[java] view plain copy

 

1. /** 
2.  *  
3.  * @author 廖钟民 
4.  * time : 2015年1月30日下午1:23:36 
5.  * @version 
6.  */  
7. public class UserPostJoin {  
8.     // 定义输入路径  
9.     private static final String INPUT_PATH1 = "hdfs://liaozhongmin:9000/user_post_join/user.txt";  
10.     private static final String INPUT_PATH2 = "hdfs://liaozhongmin:9000/user_post_join/post.txt";  
11.     // 定义输出路径  
12.     private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
13.   
14.     public static void main(String[] args) {  
15.   
16.         try {  
17.             // 创建配置信息  
18.             Configuration conf = new Configuration();  
19.   
20.             // 创建文件系统  
21.             FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
22.             // 如果输出目录存在，我们就删除  
23.             if (fileSystem.exists(new Path(OUT_PATH))) {  
24.                 fileSystem.delete(new Path(OUT_PATH), true);  
25.             }  
26.   
27.             // 创建任务  
28.             Job job = new Job(conf, UserPostJoin.class.getName());  
29.   
30.             // 设置连接类型  
31.             job.getConfiguration().set("joinType", "allOuter");  
32.             // 设置多路径输入  
33.             MultipleInputs.addInputPath(job, new Path(INPUT_PATH1), TextInputFormat.class, UserMapper.class);  
34.             MultipleInputs.addInputPath(job, new Path(INPUT_PATH2), TextInputFormat.class, PostMapper.class);  
35.   
36.             //1.2 设置自定义Mapper类和设置map函数输出数据的key和value的类型  
37.             job.setMapOutputKeyClass(Text.class);  
38.             job.setMapOutputValueClass(UserPost.class);  
39.   
40.             //1.3 设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)  
41.             job.setPartitionerClass(HashPartitioner.class);  
42.             job.setNumReduceTasks(1);  
43.   
44.             //1.4 排序  
45.             //1.5 归约  
46.             //2.1 Shuffle把数据从Map端拷贝到Reduce端。  
47.             //2.2 指定Reducer类和输出key和value的类型  
48.             job.setReducerClass(UserPostReducer.class);  
49.             job.setOutputKeyClass(Text.class);  
50.             job.setOutputValueClass(Text.class);  
51.   
52.             //2.3 指定输出的路径和设置输出的格式化类  
53.             FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
54.             job.setOutputFormatClass(TextOutputFormat.class);  
55.   
56.             // 提交作业 退出  
57.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
58.   
59.         } catch (Exception e) {  
60.             e.printStackTrace();  
61.         }  
62.     }  
63.   
64.     /** 
65.      * 自定义Mapper类用于处理来自user.txt文件的数据 
66.      * @author 廖钟民 
67.      * time : 2015年1月30日下午1:22:12 
68.      * @version 
69.      */  
70.     public static class UserMapper extends Mapper<LongWritable, Text, Text, UserPost> {  
71.         @Override  
72.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, UserPost>.Context context) throws IOException, InterruptedException {  
73.             // 对字符串进行切分  
74.             String[] arr = value.toString().split("\t");  
75.             // 创建UserId  
76.             Text userId = new Text(arr[0]);  
77.             // 把结果写出去  
78.             context.write(userId, new UserPost("U", value.toString()));  
79.         }  
80.     }  
81.     /** 
82.      * 自定义Mapper类用于处理来自post.txt文件的数据 
83.      * @author 廖钟民 
84.      * time : 2015年1月30日下午1:22:16 
85.      * @version 
86.      */  
87.     public static class PostMapper extends Mapper<LongWritable, Text, Text, UserPost> {  
88.         @Override  
89.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, UserPost>.Context context) throws IOException, InterruptedException {  
90.             // 对数据进行切分  
91.             String[] arr = value.toString().split("\t");  
92.             // 创建用户ID  
93.             Text userId = new Text(arr[0]);  
94.             context.write(userId, new UserPost("P", value.toString()));  
95.   
96.         }  
97.     }  
98.     /** 
99.      * 自定义Reducer类用于处理不同Mapper类的输出 
100.      * @author 廖钟民 
101.      * time : 2015年1月30日下午1:23:05 
102.      * @version 
103.      */  
104.     public static class UserPostReducer extends Reducer<Text, UserPost, Text, Text> {  
105.         // 定义List集合用于存放用户  
106.         private List<Text> users = new ArrayList<Text>();  
107.         private List<Text> posts = new ArrayList<Text>();  
108.   
109.         // 定义连接类型  
110.         private String joinType;  
111.   
112.         @Override  
113.         protected void setup(Reducer<Text, UserPost, Text, Text>.Context context) throws IOException, InterruptedException {  
114.   
115.             this.joinType = context.getConfiguration().get("joinType");  
116.   
117.             System.out.println(joinType);  
118.   
119.         }  
120.         /** 
121.          * 经过Shuffle后数据会分组，每一组数据都会调用一次reduce()函数 
122.          *第一组数据： 
123.          *1 lavimer 
124.          *1 1   java 
125.          *1 2   c 
126.          * 
127.          *第二组数据： 
128.          *2 3   hadoop 
129.          *2 liaozhongmin 
130.          * 
131.          *第三组数据： 
132.          *3 liaozemin 
133.          * 
134.          *第四组数据： 
135.          *4 4   hive 
136.          * 
137.          *第五组数据： 
138.          *5 5   hbase 
139.          *5 6   pig 
140.          *5 7   flume 
141.          * 
142.          *每一组数据都会调用一次reduce()函数，我们以第一组数据为例进行讲解： 
143.          * 
144.          *进入reduce函数后，<1,lavimer>会被添加到users集合中  
145.          *<1 1   java>和<1  2   c>会被添加到posts集合中 
146.          * 
147.          *然后是判断当前操作是什么类型的连接，我们以等值连接为例： 
148.          *遍历两个集合得到的数据为： 
149.          *【1    lavimer    1         1    java】 
150.          *【1    lavimer    1         2    c】 
151.          * 
152.          *这是第一组数据的执行轨迹，其他依次类推就可以得到相关的操作 
153.          */  
154.         protected void reduce(Text key, Iterable<UserPost> values, Reducer<Text, UserPost, Text, Text>.Context context) throws IOException,  
155.                 InterruptedException {  
156.             // 清空集合  
157.             users.clear();  
158.             posts.clear();  
159.   
160.             // 迭代values集合把当前穿进来的某个组中的数据分别添加到对应的集合中  
161.             for (UserPost val : values) {  
162.                 System.out.println("实际值：" + key + "===>" + values.toString());  
163.                 if (val.getType().equals("U")) {  
164.                     users.add(new Text(val.getData()));  
165.                 } else {  
166.                     posts.add(new Text(val.getData()));  
167.                 }  
168.             }  
169.   
170.             // 根据joinType关键字做对应的连接操作  
171.             if (joinType.equals("innerJoin")) {// 内连接  
172.                 if (users.size() > 0 && posts.size() > 0) {  
173.   
174.                     for (Text user : users) {  
175.                         for (Text post : posts) {  
176.                             context.write(new Text(user), new Text(post));  
177.                         }  
178.                     }  
179.                 }  
180.             } else if (joinType.equals("leftOuter")) {// 左外连接  
181.   
182.                 for (Text user : users) {  
183.                     if (posts.size() > 0) {  
184.                         for (Text post : posts) {  
185.                             context.write(new Text(user), new Text(post));  
186.                         }  
187.                     } else {  
188.                         context.write(new Text(user), createEmptyPost());  
189.                     }  
190.                 }  
191.   
192.             } else if (joinType.equals("rightOuter")) {// 右外连接  
193.                 for (Text post : posts) {  
194.                     if (users.size() > 0) {  
195.                         for (Text user : users) {  
196.                             context.write(new Text(user), new Text(post));  
197.                         }  
198.                     } else {  
199.                         context.write(createEmptyUser(), post);  
200.                     }  
201.                 }  
202.             } else if (joinType.equals("allOuter")) {// 全外连接  
203.                 if (users.size() > 0) {  
204.                     for (Text user : users) {  
205.                         if (posts.size() > 0) {  
206.                             for (Text post : posts) {  
207.                                 context.write(new Text(user), new Text(post));  
208.                             }  
209.                         } else {  
210.                             context.write(new Text(user), createEmptyPost());  
211.                         }  
212.                     }  
213.                 } else {  
214.                     for (Text post : posts) {  
215.                         if (users.size() > 0) {  
216.                             for (Text user : users) {  
217.                                 context.write(new Text(user), new Text(post));  
218.                             }  
219.                         } else {  
220.                             context.write(createEmptyUser(), post);  
221.                         }  
222.                     }  
223.                 }  
224.             }  
225.   
226.         }  
227.   
228.         /** 
229.          * 用户为空时用制表符代替 
230.          *  
231.          * @return 
232.          */  
233.         private Text createEmptyUser() {  
234.             return new Text("NULL");  
235.         }  
236.   
237.         /** 
238.          * 帖子为空时用制表符代替 
239.          *  
240.          * @return 
241.          */  
242.         private Text createEmptyPost() {  
243.             return new Text("NULL");  
244.         }  
245.     }  
246. }  
247. /** 
248.  * 自定义实体类封装两个表的数据 
249.  * @author 廖钟民 
250.  * time : 2015年1月30日下午1:23:50 
251.  * @version 
252.  */  
253. class UserPost implements Writable {  
254.   
255.     // 类型(U表示用户，P表示帖子)  
256.     private String type;  
257.     private String data;  
258.   
259.     public UserPost() {  
260.     }  
261.   
262.     public UserPost(String type, String data) {  
263.         this.type = type;  
264.         this.data = data;  
265.     }  
266.   
267.     public String getType() {  
268.         return type;  
269.     }  
270.   
271.     public void setType(String type) {  
272.         this.type = type;  
273.     }  
274.   
275.     public String getData() {  
276.         return data;  
277.     }  
278.   
279.     public void setData(String data) {  
280.         this.data = data;  
281.     }  
282.   
283.     public void write(DataOutput out) throws IOException {  
284.         out.writeUTF(this.type);  
285.         out.writeUTF(this.data);  
286.   
287.     }  
288.   
289.     public void readFields(DataInput in) throws IOException {  
290.         this.type = in.readUTF();  
291.         this.data = in.readUTF();  
292.     }  
293.   
294. }