Hadoop中的KeyValueInputFormat

Posted 2020-06-12

一：背景

有时候，我们可以不以偏移量和行文本内容来作为数据源到MapTask的输入格式，而使用键值对的形式，使用KeyValueInputFormat就可以完成这种需求。

二：技术实现

数据源如下

 

操作代码如下：

 

[java] view plain copy

 

1. public class MyKeyValueTextInputFormat {  
2.         // 定义输入路径  
3.         private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/hello";  
4.         // 定义输出路径  
5.         private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";  
6.   
7.         public static void main(String[] args) {  
8.   
9.             try {  
10.                 // 创建配置信息  
11.                 Configuration conf = new Configuration();  
12.                 //设置行的分隔符，这里是制表符，第一个制表符前面的是Key，第一个制表符后面的内容都是value  
13.                 conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t");  
14.                 /**********************************************/  
15.                 //对Map端输出进行压缩  
16.                 /*conf.setBoolean("mapred.compress.map.output", true); 
17.                 //设置map端输出使用的压缩类 
18.                 conf.setClass("mapred.map.output.compression.codec", GzipCodec.class, CompressionCodec.class); 
19.                 //对reduce端输出进行压缩 
20.                 conf.setBoolean("mapred.output.compress", true); 
21.                 //设置reduce端输出使用的压缩类 
22.                 conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);*/  
23.                 // 添加配置文件(我们可以在编程的时候动态配置信息，而不需要手动去改变集群)  
24.                 /* 
25.                  * conf.addResource("classpath://hadoop/core-site.xml");  
26.                  * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
27.                  * conf.addResource("classpath://hadoop/hdfs-site.xml"); 
28.                  */  
29.   
30.                 // 创建文件系统  
31.                 FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);  
32.                 // 如果输出目录存在，我们就删除  
33.                 if (fileSystem.exists(new Path(OUT_PATH))) {  
34.                     fileSystem.delete(new Path(OUT_PATH), true);  
35.                 }  
36.   
37.                 // 创建任务  
38.                 Job job = new Job(conf, MyKeyValueTextInputFormat.class.getName());  
39.   
40.                 //1.1   设置输入目录和设置输入数据格式化的类  
41.                 FileInputFormat.setInputPaths(job, INPUT_PATH);  
42.                 job.setInputFormatClass(KeyValueTextInputFormat.class);  
43.   
44.                 //1.2   设置自定义Mapper类和设置map函数输出数据的key和value的类型  
45.                 job.setMapperClass(MyKeyValueInputFormatMapper.class);  
46.                 job.setMapOutputKeyClass(Text.class);  
47.                 job.setMapOutputValueClass(LongWritable.class);  
48.   
49.                 //1.3   设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)  
50.                 job.setPartitionerClass(HashPartitioner.class);  
51.                 job.setNumReduceTasks(1);  
52.   
53.                 //1.4   排序、分组  
54.                 //1.5   归约  
55.                 //2.1   Shuffle把数据从Map端拷贝到Reduce端。  
56.                 //2.2   指定Reducer类和输出key和value的类型  
57.                 job.setReducerClass(MyKeyValueInputFormatReducer.class);  
58.                 job.setOutputKeyClass(Text.class);  
59.                 job.setOutputValueClass(LongWritable.class);  
60.   
61.                 //2.3   指定输出的路径和设置输出的格式化类  
62.                 FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));  
63.                 job.setOutputFormatClass(TextOutputFormat.class);  
64.   
65.                 // 提交作业 退出  
66.                 System.exit(job.waitForCompletion(true) ? 0 : 1);  
67.               
68.             } catch (Exception e) {  
69.                 e.printStackTrace();  
70.             }  
71.         }  
72.       
73.         /** 
74.          * 自定义Mapper类 
75.          * @author 廖钟民 
76.          * time : 2015年1月15日下午8:00:01 
77.          * @version 
78.          */  
79.     public static class MyKeyValueInputFormatMapper extends Mapper<Text, Text, Text, LongWritable>{  
80.   
81.         /** 
82.          * 输入数据是 
83.          * hello    you 
84.          * hello    me 
85.          * you  me  love 
86.          *  
87.          * 进入map的键值对应该是<hello,you> <hello,me> <you,me love>每个键值对分别调用map()函数 
88.          */  
89.         protected void map(Text key, Text value, Mapper<Text, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {  
90.             //把key和value都当做key写出去  
91.             context.write(key, new LongWritable(1));  
92.             context.write(value, new LongWritable(1));  
93.         }  
94.     }  
95.     /** 
96.      * map()函数的输出结果为： 
97.      *<hello,1> <you,1> <hello,1> <me,1> <you,1> <me love,1> 
98.      *排序分组后的结果为： 
99.      *<hello,{1,1}> <me,{1}> <me love,{1}> <you,{1,1}> 
100.      */  
101.     /** 
102.      * 自定义Reducer类 
103.      * @author 廖钟民 
104.      * time : 2015年1月15日下午8:00:12 
105.      * @version 
106.      */  
107.     public static class MyKeyValueInputFormatReducer extends Reducer<Text, LongWritable, Text, LongWritable>{  
108.         @Override  
109.         protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException,  
110.                 InterruptedException {  
111.               
112.             int sum = 0;  
113.             //遍历统计  
114.             for (LongWritable s : values){  
115.                 sum += s.get();  
116.             }  
117.               
118.             context.write(key, new LongWritable(sum));  
119.         }  
120.     }  
121. }  

程序运行结果：