Hadoop中的GenericWritable

Posted 2020-06-12

一：背景

某些业务的数据来源可能不同，且数据源中的分割方式也不同，导致在MapReduce编程时使用的格式化类会不同，为了包装不同的Map输出，Hadoop提供了GenericWritable类，允许我们同时操作多个不同的Map输出，输出到一个Reduce中进行处理。

 

技术实现：

我们对HDFS中两个不同的数据源进行处理，数据源如下(hello文件中的内容是通过制表符来分割的，hello2中的内容是通过逗号来分割的)：

 

为了同时处理这两个文件，我们要使用不同的Map进行处理。源码如下：

 

[java] view plain copy

 

1. public class MyGenericWritableTest {  
2.   
3.     //定义输入路径  
4.     private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/files";  
5.     //定义输出路径  
6.     private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
7.       
8.     public static void main(String[] args) {  
9.           
10.         try {  
11.             //创建配置信息  
12.             Configuration conf = new Configuration();  
13.             //创建文件系统  
14.             FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
15.             //如果输出目录存在，我们就删除  
16.             if (fileSystem.exists(new Path(OUT_PATH))){  
17.                 fileSystem.delete(new Path(OUT_PATH), true);  
18.             }  
19.               
20.             //创建任务  
21.             Job job = new Job(conf, MyGenericWritableTest.class.getName());  
22.               
23.             //1.1 设置输入目录和设置格式化的类(这是使用GenericWritable的关键)  
24.             MultipleInputs.addInputPath(job, new Path(INPUT_PATH + "/hello"), TextInputFormat.class, MyGenericMapper1.class);  
25.             MultipleInputs.addInputPath(job, new Path(INPUT_PATH + "/hello2"), TextInputFormat.class, MyGenericMapper2.class);  
26.               
27.             //1.2 设置自定义Mapper类和map函数输出数据的key和value的类型  
28.             job.setMapOutputKeyClass(Text.class);  
29.             job.setMapOutputValueClass(MyGenericWritable.class);  
30.               
31.             //1.3 设置分区和reduce的数量(reduce的数量和分区的数量对应，因为分区的数量为一个所以reduce的数量也为一个)  
32.             job.setPartitionerClass(HashPartitioner.class);  
33.             job.setNumReduceTasks(1);  
34.               
35.             //1.4 排序 分组  
36.             //1.5 归约  
37.             //2.1 Shuffle过程将map端输出的数据拷贝到reduce端  
38.               
39.             //2.2 设置自定义的Reducer类和输出的key和value类型  
40.             job.setReducerClass(MyGenericReducer.class);  
41.             job.setOutputKeyClass(Text.class);  
42.             job.setOutputValueClass(LongWritable.class);  
43.               
44.             //2.3设置输出的格式化类和指定输出路径  
45.             job.setOutputFormatClass(TextOutputFormat.class);  
46.             FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
47.               
48.             //提交作业  
49.             System.exit(job.waitForCompletion(true) ? 0 : 1);  
50.               
51.         } catch (Exception e) {  
52.             e.printStackTrace();  
53.         }  
54.       
55.           
56.     }  
57.       
58.     /** 
59.      * 自定义Mapper类用于处理hello1文件,该文件中的内容是以制表符进行分割的 
60.      * @author 廖钟民 
61.      * time : 2015年1月15日下午5:12:18 
62.      * @version 
63.      */  
64.     public static class MyGenericMapper1 extends Mapper<LongWritable, Text, Text, MyGenericWritable>{  
65.           
66.         @Override  
67.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, MyGenericWritable>.Context context) throws IOException,  
68.                 InterruptedException {  
69.                 //对value按制表符进行切分  
70.                 String[] splits = value.toString().split("\t");  
71.                 //遍历字符串数组通过context写出去  
72.                 for (String str : splits){  
73.                     context.write(new Text(str), new MyGenericWritable(new LongWritable(1)));  
74.                 }  
75.         }  
76.     }  
77.       
78.     /** 
79.      * 自定义Mapper类用于处理hello2文件，该文件的输入格式化类是TextInputFormat，用逗号给开 
80.      * @author 廖钟民 
81.      * time : 2015年1月15日下午5:13:30 
82.      * @version 
83.      */  
84.     public static class MyGenericMapper2 extends Mapper<LongWritable, Text, Text, MyGenericWritable>{  
85.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, MyGenericWritable>.Context context) throws IOException,  
86.                 InterruptedException {  
87.             //对value进行切割  
88.             String[] splits = value.toString().split(",");  
89.               
90.             //遍历字符串数组通过context写出去  
91.             for (String str : splits){  
92.                 //为了模拟不同的map端输出，我们这里故意设置一个Text类型  
93.                 context.write(new Text(str), new MyGenericWritable(new Text("1")));  
94.             }  
95.               
96.         }  
97.     }  
98.       
99.     /** 
100.      * 自定义Reducer类，对不同Map端的输出进行汇总 
101.      * @author 廖钟民 
102.      * time : 2015年1月15日下午7:09:27 
103.      * @version 
104.      */  
105.     public static class MyGenericReducer extends Reducer<Text, MyGenericWritable, Text, LongWritable>{  
106.         @Override  
107.         protected void reduce(Text key, Iterable<MyGenericWritable> values, Reducer<Text, MyGenericWritable, Text, LongWritable>.Context context)  
108.                 throws IOException, InterruptedException {  
109.             //定义单词出现的总次数  
110.             long count = 0L;  
111.             //遍历集合进行叠加  
112.             for (MyGenericWritable time : values){  
113.                 //获取MyGenericWritable对象  
114.                 Writable writable = time.get();  
115.                 //如果当前是LongWritable类型  
116.                 if (writable instanceof LongWritable){  
117.                       
118.                     count += ((LongWritable) writable).get();  
119.                 }  
120.                 //如果当前是Text类型  
121.                 if (writable instanceof Text){  
122.                     count += Long.parseLong(((Text)writable).toString());  
123.                 }  
124.             }  
125.             //把最后计算所得的结果写出去  
126.             context.write(key, new LongWritable(count));  
127.         }  
128.     }  
129. }  
130.   
131. /** 
132.  * 继承GenericWritable进行自定义 
133.  * @author 廖钟民 
134.  * time : 2015年1月15日下午4:29:50 
135.  * @version 
136.  */  
137. class MyGenericWritable extends GenericWritable{  
138.   
139.     //无参构造函数  
140.     public MyGenericWritable() {  
141.           
142.     }  
143.       
144.     //有参构造函数  
145.     public MyGenericWritable(Text text) {  
146.         super.set(text);  
147.     }  
148.       
149.     //有参构造函数  
150.     public MyGenericWritable(LongWritable longWritable) {  
151.         super.set(longWritable);  
152.     }  
153.   
154.       
155.     @Override  
156.     protected Class<? extends Writable>[] getTypes() {  
157.           
158.         return new Class[]{LongWritable.class,Text.class};  
159.     }  
160.       
161. }  

输出结果：