每日博客

11．Mapreduce实例——MapReduce自定义输出格式小

实验目的

1.了解Hadoop自带的几种输出格式

2.准确理解MapReduce自定义输出格式的设计原理

3.熟练掌握MapReduce自定义输出格式程序代码编写

4.培养自己编写MapReduce自定义输出格式程序代码解决问题的能力

实验原理

1.输出格式：提供给OutputCollector的键值对会被写到输出文件中，写入的方式由输出格式控制。OutputFormat的功能跟前面描述的InputFormat类很像，Hadoop提供的OutputFormat的实例会把文件写在本地磁盘或HDFS上。在不做设置的情况下，计算结果会以part-000*输出成多个文件，并且输出的文件数量和reduce数量一样，文件内容格式也不能随心所欲。每一个reducer会把结果输出写在公共文件夹中一个单独的文件内，这些文件的命名一般是part-nnnnn，nnnnn是关联到某个reduce任务的partition的id，输出文件夹通过FileOutputFormat.setOutputPath() 来设置。你可以通过具体MapReduce作业的JobConf对象的setOutputFormat()方法来设置具体用到的输出格式。下表给出了已提供的输出格式：

 

Hadoop提供了一些OutputFormat实例用于写入文件，基本的（默认的）实例是TextOutputFormat，它会以一行一个键值对的方式把数据写入一个文本文件里。这样后面的MapReduce任务就可以通过KeyValueInputFormat类简单的重新读取所需的输入数据了，而且也适合于人的阅读。还有一个更适合于在MapReduce作业间使用的中间格式，那就是SequenceFileOutputFormat，它可以快速的序列化任意的数据类型到文件中，而对应SequenceFileInputFormat则会把文件反序列化为相同的类型并提交为下一个Mapper的输入数据，方式和前一个Reducer的生成方式一样。NullOutputFormat不会生成输出文件并丢弃任何通过OutputCollector传递给它的键值对，如果你在要reduce()方法中显式的写你自己的输出文件并且不想Hadoop框架输出额外的空输出文件，那这个类是很有用的。

RecordWriter：这个跟InputFormat中通过RecordReader读取单个记录的实现很相似，OutputFormat类是RecordWriter对象的工厂方法，用来把单个的记录写到文件中，就像是OuputFormat直接写入的一样。

2.与IntputFormat相似， 当面对一些特殊情况时，如想要Reduce支持多个输出，这时Hadoop本身提供的TextOutputFormat、SequenceFileOutputFormat、NullOutputFormat等肯定是无法满足我们的需求，这时我们需要自定义输出数据格式。类似输入数据格式，自定义输出数据格式同样可以参考下面的步骤：

（1） 自定义一个继承OutputFormat的类，不过一般继承FileOutputFormat即可；

（2）实现其getRecordWriter方法，返回一个RecordWriter类型；

（3）自定义一个继承RecordWriter的类，定义其write方法，针对每个<key,value>写入文件数据；

实验环境

Linux Ubuntu 14.04

jdk-7u75-linux-x64

hadoop-2.6.0-cdh5.4.5

hadoop-2.6.0-eclipse-cdh5.4.5.jar

eclipse-java-juno-SR2-linux-gtk-x86_64

 

 

实验内容

当面对一些特殊的<key,value>键值对时，要求开发人员继承FileOutputFormat，用于实现一种新的输出格式。同时还需继承RecordWriter，用于实现新输出格式key和value的写入方法。现在我们有某电商数据表cat_group1，包含（分组id，分组名称，分组码，奢侈品标记）四个字段cat_group1的数据内容如下：

cat_group1（group_id，group_name，group_code，flag）

1. 分组id 分组名称 分组码 奢侈品标记  
2. 512     奢侈品    c        1  
3. 675     箱包      1       1  
4. 676     化妆品    2        1  
5. 677     家电        3     1  
6. 501     有机食品      1     0  
7. 502     蔬菜水果      2     0  
8. 503     肉禽蛋奶      3     0  
9. 504     深海水产      4     0  

10. 505     地方特产      5     0  

11. 506     进口食品      6     0  

要求把相同奢侈品标记（flag）的数据放入到一个文件里，并且以该字段来命名文件的名称，输出时key与value 以“：”分割，形如"key：value"

结果输出0.txt和1.txt两文件。

0.txt

1. 奢侈品标记：分组ID 分组名称 分组码  
2.         0:506     进口食品  6  
3.         0:505     地方特产  5  
4.         0:504     深海水产  4  
5.         0:503     肉禽蛋奶  3  
6.         0:502     蔬菜水果  2  
7.         0:501     有机食品  1  

1.txt

1. 奢侈品标记：分组ID 分组名称 分组码  
2.         1:677      家电     3  
3.         1:676     化妆品    2  
4.         1:675      箱包     1  
5.         1:512     奢侈品    c  

实验步骤

1.切换到/apps/hadoop/sbin目录下，开启Hadoop。

1. cd /apps/hadoop/sbin  
2. ./start-all.sh  

2.在Linux本地新建/data/mapreduce12目录。

1. mkdir -p /data/mapreduce12  

3.在Linux中切换到/data/mapreduce12目录下，用wget命令从http://192.168.1.100:60000/allfiles/mapreduce12/cat_group1网址上下载文本文件cat_group1。

1. cd /data/mapreduce12  
2. wget http://192.168.1.100:60000/allfiles/mapreduce12/cat_group1  

然后在当前目录下用wget命令从http://192.168.1.100:60000/allfiles/mapreduce12/hadoop2lib.tar.gz网址上下载项目用到的依赖包。

1. wget http://192.168.1.100:60000/allfiles/mapreduce12/hadoop2lib.tar.gz  

将hadoop2lib.tar.gz解压到当前目录下。

1. tar zxvf hadoop2lib.tar.gz  

4.首先在HDFS上新建/mymapreduce12/in目录，然后将Linux本地/data/mapreduce12目录下的cat_group1文件导入到HDFS的/mymapreduce12/in目录中。

1. hadoop fs -mkdir -p /mymapreduce12/in  
2. hadoop fs -put /data/mapreduce12/cat_group1 /mymapreduce12/in  

5.新建Java Project项目，项目名为mapreduce12。

 

 

在mapreduce12项目下新建包，包名为mapreduce。

 

 

在mapredcue包下新建名为MyMultipleOutputFormat的类。

 

 

在mapredcue包下新建名为FileOutputMR的类。

 

6.添加项目所需依赖的jar包，右键单击项目名，新建一个文件夹hadoop2lib，用于存放项目所需的jar包。

 

 

将/data/mapreduce12目录下，hadoop2lib目录中的jar包，拷贝到eclipse中mapreduce12项目的hadoop2lib目录下。

 

选中所有项目hadoop2lib目录下所有jar包，并添加到Build Path中。

 

7.编写程序代码，并描述设计思路

自定义FileRecordWriter类命名为MyMultipleOutputFormat，它继承了FileRecordWriter类，并且它里面主要包含三部分：类中的getRecordWriter、getTaskOutputPath、generateFileNameForKayValue方法和两个内部类LineRecordWriter、MutiRecordWriter

类中的方法代码：

1. private MultiRecordWriter writer=null;  
2.     public RecordWriter<K,V> getRecordWriter(TaskAttemptContext job) throws IOException{  
3.         if(writer==null){  
4.             writer=new MultiRecordWriter(job,getTaskOutputPath(job));  
5.         }  
6.         return writer;  
7.     }  
8.     private Path getTaskOutputPath(TaskAttemptContext conf) throws IOException{  
9.         Path workPath=null;  
10. 10.         OutputCommitter committer=super.getOutputCommitter(conf);  
11. 11.         if(committer instanceof FileOutputCommitter){  
12. 12.             workPath=((FileOutputCommitter) committer).getWorkPath();  
13. 13.         }else{  
14. 14.             Path outputPath=super.getOutputPath(conf);  
15. 15.             if(outputPath==null){  
16. 16.                 throw new IOException("Undefined job output-path");  
17. 17.             }  
18. 18.             workPath=outputPath;  
19. 19.         }  
20. 20.         return workPath;  
21. 21.     }  
22. 22.     protected abstract String generateFileNameForKayValue(K key,V value,Configuration conf);  

getRecordWriter()方法判断该类实例是否存在，若不存在则创建一个实例。getTaskOutputPath()方法获取工作任务的输出路径。generateFileNameForKayValue()方法是抽象的，通过key、value 和conf三个参数确定key/value输出的文件名，并将其返回。

LineRecordWriter类代码：

1. protected static class LineRecordWriter<K,V> extends RecordWriter<K, V> {  
2.         private static final String utf8 = "UTF-8";  
3.         private static final byte[] newline;  
4.         private PrintWriter tt;  
5.         static {  
6.           try {  
7.             newline = "\n".getBytes(utf8);  
8.           } catch (UnsupportedEncodingException uee) {  
9.             throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
10. 10.           }  
11. 11.         }  
12. 12.   
13. 13.         protected DataOutputStream out;  
14. 14.         private final byte[] keyValueSeparator;  
15. 15.   
16. 16.         public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {  
17. 17.           this.out = out;  
18. 18.           try {  
19. 19.             this.keyValueSeparator = keyValueSeparator.getBytes(utf8);  
20. 20.           } catch (UnsupportedEncodingException uee) {  
21. 21.             throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
22. 22.           }  
23. 23.         }  
24. 24.   
25. 25.         public LineRecordWriter(DataOutputStream out) {  
26. 26.           this(out, ":");  
27. 27.         }  
28. 28.         private void writeObject(Object o) throws IOException {  
29. 29.           if (o instanceof Text) {  
30. 30.             Text to = (Text) o;  
31. 31.             out.write(to.getBytes(), 0, to.getLength());  
32. 32.           } else {  
33. 33.             out.write(o.toString().getBytes(utf8));  
34. 34.           }  
35. 35.         }  
36. 36.   
37. 37.         public synchronized void write(K key, V value)  
38. 38.           throws IOException {  
39. 39.           boolean nullKey = key == null || key instanceof NullWritable;  
40. 40.           boolean nullValue = value == null || value instanceof NullWritable;  
41. 41.           if (nullKey && nullValue) {//  
42. 42.             return;  
43. 43.           }  
44. 44.           if (!nullKey) {  
45. 45.             writeObject(key);  
46. 46.           }  
47. 47.           if (!(nullKey || nullValue)) {  
48. 48.             out.write(keyValueSeparator);  
49. 49.           }  
50. 50.           if (!nullValue) {  
51. 51.             writeObject(value);  
52. 52.           }  
53. 53.           out.write(newline);  
54. 54.   
55. 55.         }  
56. 56.         public synchronized  
57. 57.         void close(TaskAttemptContext context) throws IOException {  
58. 58.           out.close();  
59. 59.         }  
60. 60.       }  

LineRecordWriter类主要是为<key,value>输出时定义它的输出格式。通过加线程同步关键字 synchronized对write()方法上锁。write()方法首先从输出流中写入key-value，然后判断键值对是否为空，如果k-v为空，则操作失败返回空，如果key不为空，则写入key，如果key，value 都不为空则，在中间写入k-v分隔符，如果value不为空，则写入value，最后写入换行符。

MutiRecordWriter类代码：

1. public class MultiRecordWriter extends RecordWriter<K,V>{  
2.         private HashMap<String,RecordWriter<K,V> >recordWriters=null;  
3.         private TaskAttemptContext job=null;  
4.         private Path workPath=null;  
5.         public MultiRecordWriter(TaskAttemptContext job,Path workPath){  
6.             super();  
7.             this.job=job;  
8.             this.workPath=workPath;  
9.             recordWriters=new HashMap<String,RecordWriter<K,V>>();  
10. 10.   
11. 11.         }  
12. 12.         public void close(TaskAttemptContext context) throws IOException, InterruptedException{  
13. 13.             Iterator<RecordWriter<K,V>> values=this.recordWriters.values().iterator();  
14. 14.             while(values.hasNext()){  
15. 15.                 values.next().close(context);  
16. 16.             }  
17. 17.             this.recordWriters.clear();  
18. 18.         }  
19. 19.         public void write(K key,V value) throws IOException, InterruptedException{  
20. 20.             String baseName=generateFileNameForKayValue(key ,value,job.getConfiguration());  
21. 21.             RecordWriter<K,V> rw=this.recordWriters.get(baseName);  
22. 22.             if(rw==null){  
23. 23.                 rw=getBaseRecordWriter(job,baseName);  
24. 24.                 this.recordWriters.put(baseName,rw);  
25. 25.             }  
26. 26.             rw.write(key, value);  
27. 27.         }  
28. 28.   
29. 29.   
30. 30.         private RecordWriter<K,V> getBaseRecordWriter(TaskAttemptContext job,String baseName)throws IOException,InterruptedException{  
31. 31.             Configuration conf=job.getConfiguration();  
32. 32.             boolean isCompressed=getCompressOutput(job);  
33. 33.             String keyValueSeparator= ":";  
34. 34.             RecordWriter<K,V> recordWriter=null;  
35. 35.             if(isCompressed){  
36. 36.                 Class<? extends CompressionCodec> codecClass=getOutputCompressorClass(job,(Class<? extends CompressionCodec>) GzipCodec.class);  
37. 37.                 CompressionCodec codec=ReflectionUtils.newInstance(codecClass,conf);  
38. 38.                 Path file=new Path(workPath,baseName+codec.getDefaultExtension());  
39. 39.                 FSDataOutputStream fileOut=file.getFileSystem(conf).create(file,false);  
40. 40.                 recordWriter=new LineRecordWriter<K,V>(new DataOutputStream(codec.createOutputStream(fileOut)),keyValueSeparator);  
41. 41.             }else{  
42. 42.                 Path file=new Path(workPath,baseName);  
43. 43.                 FSDataOutputStream fileOut=file.getFileSystem(conf).create(file,false);  
44. 44.                 recordWriter =new LineRecordWriter<K,V>(fileOut,keyValueSeparator);  
45. 45.             }  
46. 46.             return recordWriter;  
47. 47.         }  
48. 48.     }  

write()方法得到输出的文件名0.txt和1.txt并将两文件写到hdfs上，close()方法关闭输出文件的数据流。getBaseRecordWriter()方法首先用getCompressOutput(job) 从配置判断输出是否压缩，根据是否压缩获取相应的LineRecordWriter。

MyMultipleOutputFormat完整代码：

1. package mapreduce;  
2. import java.io.DataOutputStream;  
3. import java.io.IOException;  
4. import java.io.PrintWriter;  
5. import java.io.UnsupportedEncodingException;  
6. import java.util.HashMap;  
7. import java.util.Iterator;  
8. import org.apache.hadoop.conf.Configuration;  
9. import org.apache.hadoop.fs.FSDataOutputStream;  

10. import org.apache.hadoop.fs.Path;  

11. import org.apache.hadoop.io.NullWritable;  

12. import org.apache.hadoop.io.Text;  

13. import org.apache.hadoop.io.Writable;  

14. import org.apache.hadoop.io.WritableComparable;  

15. import org.apache.hadoop.io.compress.CompressionCodec;  

16. import org.apache.hadoop.io.compress.GzipCodec;  

17. import org.apache.hadoop.mapreduce.OutputCommitter;  

18. import org.apache.hadoop.mapreduce.RecordWriter;  

19. import org.apache.hadoop.mapreduce.TaskAttemptContext;  

20. import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;  

21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  

22. import org.apache.hadoop.util.ReflectionUtils;  

23. public abstract class MyMultipleOutputFormat <K extends WritableComparable<?>,V extends Writable> extends FileOutputFormat<K,V>{  

1. 24.     private MultiRecordWriter writer=null;  
2. 25.     public RecordWriter<K,V> getRecordWriter(TaskAttemptContext job) throws IOException{  
3. 26.         if(writer==null){  
4. 27.             writer=new MultiRecordWriter(job,getTaskOutputPath(job));  
5. 28.         }  
6. 29.         return writer;  
7. 30.     }  
8. 31.     private Path getTaskOutputPath(TaskAttemptContext conf) throws IOException{  
9. 32.         Path workPath=null;  
10. 33.         OutputCommitter committer=super.getOutputCommitter(conf);  
11. 34.         if(committer instanceof FileOutputCommitter){  
12. 35.             workPath=((FileOutputCommitter) committer).getWorkPath();  
13. 36.         }else{  
14. 37.             Path outputPath=super.getOutputPath(conf);  
15. 38.             if(outputPath==null){  
16. 39.                 throw new IOException("Undefined job output-path");  
17. 40.             }  
18. 41.             workPath=outputPath;  
19. 42.         }  
20. 43.         return workPath;  
21. 44.     }  
22. 45.     protected abstract String generateFileNameForKayValue(K key,V value,Configuration conf);  
23. 46.     protected static class LineRecordWriter<K,V> extends RecordWriter<K, V> {  
24. 47.         private static final String utf8 = "UTF-8";  
25. 48.         private static final byte[] newline;  
26. 49.         private PrintWriter tt;  
27. 50.         static {  
28. 51.           try {  
29. 52.             newline = "\n".getBytes(utf8);  
30. 53.           } catch (UnsupportedEncodingException uee) {  
31. 54.             throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
32. 55.           }  
33. 56.         }  
34. 57.   
35. 58.         protected DataOutputStream out;  
36. 59.         private final byte[] keyValueSeparator;  
37. 60.   
38. 61.         public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {  
39. 62.           this.out = out;  
40. 63.           try {  
41. 64.             this.keyValueSeparator = keyValueSeparator.getBytes(utf8);  
42. 65.           } catch (UnsupportedEncodingException uee) {  
43. 66.             throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
44. 67.           }  
45. 68.         }  
46. 69.   
47. 70.         public LineRecordWriter(DataOutputStream out) {  
48. 71.           this(out, ":");  
49. 72.         }  
50. 73.         private void writeObject(Object o) throws IOException {  
51. 74.           if (o instanceof Text) {  
52. 75.             Text to = (Text) o;  
53. 76.             out.write(to.getBytes(), 0, to.getLength());  
54. 77.           } else {  
55. 78.             out.write(o.toString().getBytes(utf8));  
56. 79.           }  
57. 80.         }  
58. 81.   
59. 82.         public synchronized void write(K key, V value)  
60. 83.           throws IOException {  
61. 84.           boolean nullKey = key == null || key instanceof NullWritable;  
62. 85.           boolean nullValue = value == null || value instanceof NullWritable;  
63. 86.           if (nullKey && nullValue) {//  
64. 87.             return;  
65. 88.           }  
66. 89.           if (!nullKey) {  
67. 90.             writeObject(key);  
68. 91.           }  
69. 92.           if (!(nullKey || nullValue)) {  
70. 93.             out.write(keyValueSeparator);  
71. 94.           }  
72. 95.           if (!nullValue) {  
73. 96.             writeObject(value);  
74. 97.           }  
75. 98.           out.write(newline);  
76. 99.   
77.         }  
78.   
79.         public synchronized  
80.         void close(TaskAttemptContext context) throws IOException {  
81.           out.close();  
82.         }  
83.       }  
84.     public class MultiRecordWriter extends RecordWriter<K,V>{  
85.         private HashMap<String,RecordWriter<K,V> >recordWriters=null;  
86.         private TaskAttemptContext job=null;  
87.         private Path workPath=null;  
88.         public MultiRecordWriter(TaskAttemptContext job,Path workPath){  
89.             super();  
90.             this.job=job;  
91.             this.workPath=workPath;  
92.             recordWriters=new HashMap<String,RecordWriter<K,V>>();  
93.   
94.         }  
95.         public void close(TaskAttemptContext context) throws IOException, InterruptedException{  
96.             Iterator<RecordWriter<K,V>> values=this.recordWriters.values().iterator();  
97.             while(values.hasNext()){  
98.                 values.next().close(context);  
99.             }  
100.             this.recordWriters.clear();  
101.         }  
102.         public void write(K key,V value) throws IOException, InterruptedException{  
103.             String baseName=generateFileNameForKayValue(key ,value,job.getConfiguration());  
104.             RecordWriter<K,V> rw=this.recordWriters.get(baseName);  
105.             if(rw==null){  
106.                 rw=getBaseRecordWriter(job,baseName);  
107.                 this.recordWriters.put(baseName,rw);  
108.             }  
109.             rw.write(key, value);  
110.         }  
111.   
112.   
113.         private RecordWriter<K,V> getBaseRecordWriter(TaskAttemptContext job,String baseName)throws IOException,InterruptedException{  
114.             Configuration conf=job.getConfiguration();  
115.             boolean isCompressed=getCompressOutput(job);  
116.             String keyValueSeparator= ":";  
117.             RecordWriter<K,V> recordWriter=null;  
118.             if(isCompressed){  
119.                 Class<?extends CompressionCodec> codecClass=getOutputCompressorClass(job,(Class<?extends CompressionCodec>) GzipCodec.class);  
120.                 CompressionCodec codec=ReflectionUtils.newInstance(codecClass,conf);  
121.                 Path file=new Path(workPath,baseName+codec.getDefaultExtension());  
122.                 FSDataOutputStream fileOut=file.getFileSystem(conf).create(file,false);  
123.                 recordWriter=new LineRecordWriter<K,V>(new DataOutputStream(codec.createOutputStream(fileOut)),keyValueSeparator);  
124.             }else{  
125.                 Path file=new Path(workPath,baseName);  
126.                 FSDataOutputStream fileOut=file.getFileSystem(conf).create(file,false);  
127.                 recordWriter =new LineRecordWriter<K,V>(fileOut,keyValueSeparator);  
128.             }  
129.             return recordWriter;  
130.         }  
131.     }  
132. }  

测试程序代码也分为三部分Mapper部分reducer部分还有在里面添加一个静态类AlphabetOutputFormat。另外要注意在主函数里面把job的输出格式类设置为AlphabetOutputFormat类。

Mapper代码：

1. public static class TokenizerMapper extends Mapper<Object,Text,Text,Text>{  
2.         private Text val=new Text();  
3.         public void map(Object key,Text value,Context context)throws IOException,InterruptedException{  
4.             String str[]=value.toString().split("\t");  
5.             val.set(str[0]+" "+str[1]+" "+str[2]);  
6.                 context.write(new Text(str[3]), val);  
7.         }  
8.     }  

用split("\t")把数据截取出来，把代表flag的字段作为key,剩下的字段作为value，用context的write()方法将<key,value>直接输出。

reducer代码：

1. public static class IntSumReducer extends Reducer<Text,Text,Text,Text>{  
2.         public void reduce(Text key,Iterable<Text> values,Context context)  
3.     throws IOException,InterruptedException{  
4.     for(Text val:values){  
5.     context.write(key,val);  
6.     }  
7.     }  
8.     }  

map输出的<key,value>键值对先经过shuffle，把key相同的value值放到一个迭代器中形成values，在将<key,values>传递给reduce函数，reduce函数将输入的key直接复制给输出的key，将输入的values通过增强版for循环遍历，并把里面的每个元素赋值给输出的value，再用context的write()方法进行逐一输出<key,value>，输出的次数为循环的次数。

AlphabetOutputFormat代码：

1. public static class AlphabetOutputFormat extends MyMultipleOutputFormat<Text,Text>{  
2.         protected String generateFileNameForKayValue(Text key,Text value,Configuration conf){  
3.             return key+".txt";  
4.         }  
5.     }  

该类继承MyMultipleOutputFormat类并重写generateFileNameForKayValue()抽象方法，令其返回值为key+".txt"。

测试类完整代码：

1. package mapreduce;  
2. import java.io.IOException;  
3. import org.apache.hadoop.conf.Configuration;  
4. import org.apache.hadoop.fs.Path;  
5. import org.apache.hadoop.io.Text;  
6. import org.apache.hadoop.mapreduce.Job;  
7. import org.apache.hadoop.mapreduce.Mapper;  
8. import org.apache.hadoop.mapreduce.Reducer;  
9. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  

10. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  

11. public class FileOutputMR {  

1. 12.     public static class TokenizerMapper extends Mapper<Object,Text,Text,Text>{  
2. 13.         private Text val=new Text();  
3. 14.         public void map(Object key,Text value,Context context)throws IOException,InterruptedException{  
4. 15.             String str[]=value.toString().split("\t");  
5. 16.             val.set(str[0]+" "+str[1]+" "+str[2]);  
6. 17.                 context.write(new Text(str[3]), val);  
7. 18.         }  
8. 19.     }  
9. 20.     public static class IntSumReducer extends Reducer<Text,Text,Text,Text>{  
10. 21.         public void reduce(Text key,Iterable<Text> values,Context context)  
11. 22.     throws IOException,InterruptedException{  
12. 23.     for(Text val:values){  
13. 24.     context.write(key,val);  
14. 25.     }  
15. 26.     }  
16. 27.     }  
17. 28.     public static class AlphabetOutputFormat extends MyMultipleOutputFormat<Text,Text>{  
18. 29.     protected String generateFileNameForKayValue(Text key,Text value,Configuration conf){  
19. 30.     return key+".txt";  
20. 31.     }  
21. 32.     }  
22. 33.     public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException{  
23. 34.     Configuration conf=new Configuration();  
24. 35.     Job job=new Job(conf,"FileOutputMR");  
25. 36.     job.setJarByClass(FileOutputMR.class);  
26. 37.     job.setMapperClass(TokenizerMapper.class);  
27. 38.     job.setCombinerClass(IntSumReducer.class);  
28. 39.     job.setReducerClass(IntSumReducer.class);  
29. 40.     job.setOutputKeyClass(Text.class);  
30. 41.     job.setOutputValueClass(Text.class);  
31. 42.     job.setOutputFormatClass(AlphabetOutputFormat.class);  
32. 43.     FileInputFormat.addInputPath(job,new Path("hdfs://localhost:9000/mymapreduce12/in/cat_group1"));  
33. 44.     FileOutputFormat.setOutputPath(job,new Path("hdfs://localhost:9000/mymapreduce12/out"));  
34. 45.     System.exit(job.waitForCompletion(true)?0:1);  
35. 46.     }  
36. 47.     }  

8.在FileOutputMR类文件中，右键并点击=>Run As=>Run on Hadoop选项，将MapReduce任务提交到Hadoop中。

 

9.待执行完毕后，进入命令模式，在HDFS上从mymapreduce12/out中查看实验结果。

1. hadoop fs -ls /mymapreduce12/out  
2. hadoop fs -cat /mymapreduce12/out/0.txt  
3. hadoop fs -cat /mymapreduce12/out/1.txt