Hadoop中的KeyValueInputFormat

一：背景

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

程序运行结果：