hadoop1.2.1 MultipleOutputs将结果输出到多个文件或文件夹

时间：2022-05-12 23:53:00

hadoop1.2.1 MultipleOutputs将结果输出到多个文件或文件夹

博客分类：http://tydldd.iteye.com/blog/2053867

 

hadoop1.2.1中使用MultipleOutputs将结果输出到多个文件或文件夹

使用步骤主要有三步：

1、在reduce或map类中创建MultipleOutputs对象，将结果输出

  1. class reduceStatistics extends Reducer<Text, IntWritable, Text, IntWritable>{
  2. //将结果输出到多个文件或多个文件夹
  3. private MultipleOutputs<Text,IntWritable> mos;
  4. //创建对象
  5. protected void setup(Context context) throws IOException,InterruptedException {
  6. mos = new MultipleOutputs<Text, IntWritable>(context);
  7. }
  8. //关闭对象
  9. protected void cleanup(Context context) throws IOException,InterruptedException {
  10. mos.close();
  11. }
  12. }

2、在map或reduce方法中使用MultipleOutputs对象输出数据，代替congtext.write()

  1. protected void reduce(Text key, Iterable<IntWritable> values, Context context)
  2. throws IOException, InterruptedException {
  3. IntWritable V = new IntWritable();
  4. int sum = 0;
  5. for(IntWritable value : values){
  6. sum = sum + value.get();
  7. }
  8. System.out.println("word:" + key.toString() + "     sum = " + sum);
  9. V.set(sum);
  10. //使用MultipleOutputs对象输出数据
  11. if(key.toString().equals("hello")){
  12. mos.write("hello", key, V);
  13. }else if(key.toString().equals("world")){
  14. mos.write("world", key, V);
  15. }else if(key.toString().equals("hadoop")){
  16. //输出到hadoop/hadoopfile-r-00000文件
  17. mos.write("hadoopfile", key, V, "hadoop/");
  18. }
  19. }

3、在创建job时，定义附加的输出文件，这里的文件名称与第二步设置的文件名相同

  1. //定义附加的输出文件
  2. MultipleOutputs.addNamedOutput(job,"hello",TextOutputFormat.class,Text.class,IntWritable.class);
  3. MultipleOutputs.addNamedOutput(job,"world",TextOutputFormat.class,Text.class,IntWritable.class);
  4. MultipleOutputs.addNamedOutput(job,"hadoopfile",TextOutputFormat.class,Text.class,IntWritable.class);

完整代码：

  1. package com.ru.hadoop.wordcount;
  2. import java.io.IOException;
  3. import java.net.URI;
  4. import java.net.URISyntaxException;
  5. import java.util.regex.Pattern;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.conf.Configured;
  8. import org.apache.hadoop.fs.FileSystem;
  9. import org.apache.hadoop.fs.Path;
  10. import org.apache.hadoop.io.IntWritable;
  11. import org.apache.hadoop.io.LongWritable;
  12. import org.apache.hadoop.io.NullWritable;
  13. import org.apache.hadoop.io.Text;
  14. import org.apache.hadoop.mapred.JobConf;
  15. import org.apache.hadoop.mapred.RecordWriter;
  16. import org.apache.hadoop.mapred.lib.MultipleOutputFormat;
  17. import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;
  18. import org.apache.hadoop.mapreduce.Job;
  19. import org.apache.hadoop.mapreduce.Mapper;
  20. import org.apache.hadoop.mapreduce.Reducer;
  21. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  22. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  23. import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
  24. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  25. import org.apache.hadoop.util.Progressable;
  26. public class WordCount2 extends Configured{
  27. public static void main(String[] args) {
  28. String in = "/home/nange/work/test/word/";
  29. String out = "hdfs://localhost:9000/hdfs/test/wordcount/out/";
  30. Job job;
  31. try {
  32. //删除hdfs目录
  33. WordCount2 wc2 = new WordCount2();
  34. wc2.removeDir(out);
  35. job = new Job(new Configuration(), "wordcount Job");
  36. job.setOutputKeyClass(Text.class);
  37. job.setOutputValueClass(IntWritable.class);
  38. job.setMapperClass(mapperString.class);
  39. //          job.setCombinerClass(reduceStatistics.class);
  40. job.setReducerClass(reduceStatistics.class);
  41. //定义附加的输出文件
  42. MultipleOutputs.addNamedOutput(job,"hello",TextOutputFormat.class,Text.class,IntWritable.class);
  43. MultipleOutputs.addNamedOutput(job,"world",TextOutputFormat.class,Text.class,IntWritable.class);
  44. MultipleOutputs.addNamedOutput(job,"hadoopfile",TextOutputFormat.class,Text.class,IntWritable.class);
  45. FileInputFormat.addInputPath(job, new Path(in));
  46. FileOutputFormat.setOutputPath(job, new Path(out));
  47. job.waitForCompletion(true);
  48. } catch (IOException e) {
  49. e.printStackTrace();
  50. } catch (URISyntaxException e) {
  51. e.printStackTrace();
  52. } catch (ClassNotFoundException e) {
  53. e.printStackTrace();
  54. } catch (InterruptedException e) {
  55. e.printStackTrace();
  56. }
  57. }
  58. public void removeDir(String filePath) throws IOException, URISyntaxException{
  59. String url = "hdfs://localhost:9000";
  60. FileSystem fs  = FileSystem.get(new URI(url), new Configuration());
  61. fs.delete(new Path(filePath));
  62. }
  63. }
  64. /**
  65. * 重写maptask使用的map方法
  66. * @author nange
  67. *
  68. */
  69. class mapperString extends Mapper<LongWritable, Text, Text, IntWritable>{
  70. //设置正则表达式的编译表达形式
  71. public static Pattern PATTERN = Pattern.compile(" ");
  72. Text K = new Text();
  73. IntWritable V = new IntWritable(1);
  74. @Override
  75. protected void map(LongWritable key, Text value, Context context)
  76. throws IOException, InterruptedException {
  77. String[] words = PATTERN.split(value.toString());
  78. System.out.println("********" + value.toString());
  79. for(String word : words){
  80. K.set(word);
  81. context.write(K, V);
  82. }
  83. }
  84. }
  85. /**
  86. * 对单词做统计
  87. * @author nange
  88. *
  89. */
  90. class reduceStatistics extends Reducer<Text, IntWritable, Text, IntWritable>{
  91. //将结果输出到多个文件或多个文件夹
  92. private MultipleOutputs<Text,IntWritable> mos;
  93. //创建MultipleOutputs对象
  94. protected void setup(Context context) throws IOException,InterruptedException {
  95. mos = new MultipleOutputs<Text, IntWritable>(context);
  96. }
  97. @Override
  98. protected void reduce(Text key, Iterable<IntWritable> values, Context context)
  99. throws IOException, InterruptedException {
  100. IntWritable V = new IntWritable();
  101. int sum = 0;
  102. for(IntWritable value : values){
  103. sum = sum + value.get();
  104. }
  105. System.out.println("word:" + key.toString() + "     sum = " + sum);
  106. V.set(sum);
  107. //使用MultipleOutputs对象输出数据
  108. if(key.toString().equals("hello")){
  109. mos.write("hello", key, V);
  110. }else if(key.toString().equals("world")){
  111. mos.write("world", key, V);
  112. }else if(key.toString().equals("hadoop")){
  113. //输出到hadoop/hadoopfile-r-00000文件
  114. mos.write("hadoopfile", key, V, "hadoop/");
  115. }
  116. }
  117. //关闭MultipleOutputs对象
  118. protected void cleanup(Context context) throws IOException,InterruptedException {
  119. mos.close();
  120. }
  121. }

相关文章