Hadoop中的多输出格式

jk9hmnmh  于 2022-12-11  发布在  Hadoop
关注(0)|答案(4)|浏览(342)

我是Hadoop的新手。我正在试用Wordcount程序。
现在，为了测试多个输出文件，我使用了MultipleOutputFormat。这个链接帮助我做到了这一点。http://hadoop.apache.org/common/docs/r0.19.0/api/org/apache/hadoop/mapred/lib/MultipleOutputs.html
在我的驾驶课上

  1. MultipleOutputs.addNamedOutput(conf, "even",
  2.             org.apache.hadoop.mapred.TextOutputFormat.class, Text.class,
  3.             IntWritable.class);
  5.     MultipleOutputs.addNamedOutput(conf, "odd",
  6.             org.apache.hadoop.mapred.TextOutputFormat.class, Text.class,
  7.             IntWritable.class);`

而我被精简班级变成了这个

  1. public static class Reduce extends MapReduceBase implements
  2.         Reducer<Text, IntWritable, Text, IntWritable> {
  3.     MultipleOutputs mos = null;
  5.     public void configure(JobConf job) {
  6.         mos = new MultipleOutputs(job);
  7.     }
  9.     public void reduce(Text key, Iterator<IntWritable> values,
  10.             OutputCollector<Text, IntWritable> output, Reporter reporter)
  11.             throws IOException {
  12.         int sum = 0;
  13.         while (values.hasNext()) {
  14.             sum += values.next().get();
  15.         }
  16.         if (sum % 2 == 0) {
  17.             mos.getCollector("even", reporter).collect(key, new IntWritable(sum));
  18.         }else {
  19.             mos.getCollector("odd", reporter).collect(key, new IntWritable(sum));
  20.         }
  21.         //output.collect(key, new IntWritable(sum));
  22.     }
  23.     @Override
  24.     public void close() throws IOException {
  25.         // TODO Auto-generated method stub
  26.     mos.close();
  27.     }
  28. }

事情的工作，但我得到很多文件，（一个奇数和一个偶数为每个Map减少）
问题是：我怎么能只有2个输出文件（奇数和偶数），使每一个奇数输出的每一个Map减少得到写入到奇数文件，并为偶数相同。

hadoop

4条答案

按热度按时间
czq61nw1

czq61nw11#

每个Reducer都使用OutputFormat来写入记录。这就是为什么每个Reducer都有一组奇数和偶数文件的原因。这是设计好的，这样每个Reducer都可以并行执行写入操作。
如果只需要一个奇数和一个偶数文件，则需要将mapred.reduce.tasks设置为1。但是性能会受到影响，因为所有Map器都将输入到一个reducer中。
另一个选择是更改读取这些文件的进程以接受多个输入文件，或者编写一个单独的进程将这些文件合并在一起。

赞(0）分享  回复(0）举报  2022-12-11
dvtswwa3

dvtswwa32#

我为此写了一个类。用它来做你的工作：

  1. job.setOutputFormatClass(m_customOutputFormatClass);

这是我的课：

  1. import java.io.IOException;
  2. import java.util.HashMap;
  3. import java.util.Map;
  4. import java.util.Map.Entry;
  6. import org.apache.hadoop.fs.Path;
  7. import org.apache.hadoop.mapreduce.RecordWriter;
  8. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  9. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  11. /**
  12.  * TextOutputFormat extension which enables writing the mapper/reducer's output in multiple files.<br>
  13.  * <p>
  14.  * <b>WARNING</b>: The number of different folder shuoldn't be large for one mapper since we keep an
  15.  * {@link RecordWriter} instance per folder name.
  16.  * </p>
  17.  * <p>
  18.  * In this class the folder name is defined by the written entry's key.<br>
  19.  * To change this behavior simply extend this class and override the
  20.  * {@link HdMultipleFileOutputFormat#getFolderNameExtractor()} method and create your own
  21.  * {@link FolderNameExtractor} implementation.
  22.  * </p>
  23.  * 
  24.  * 
  25.  * @author ykesten
  26.  * 
  27.  * @param <K> - Keys type
  28.  * @param <V> - Values type
  29.  */
  30. public class HdMultipleFileOutputFormat<K, V> extends TextOutputFormat<K, V> {
  32.     private String folderName;
  34.     private class MultipleFilesRecordWriter extends RecordWriter<K, V> {
  36.         private Map<String, RecordWriter<K, V>> fileNameToWriter;
  37.         private FolderNameExtractor<K, V> fileNameExtractor;
  38.         private TaskAttemptContext job;
  40.         public MultipleFilesRecordWriter(FolderNameExtractor<K, V> fileNameExtractor, TaskAttemptContext job) {
  41.             fileNameToWriter = new HashMap<String, RecordWriter<K, V>>();
  42.             this.fileNameExtractor = fileNameExtractor;
  43.             this.job = job;
  44.         }
  46.         @Override
  47.         public void write(K key, V value) throws IOException, InterruptedException {
  48.             String fileName = fileNameExtractor.extractFolderName(key, value);
  49.             RecordWriter<K, V> writer = fileNameToWriter.get(fileName);
  50.             if (writer == null) {
  51.                 writer = createNewWriter(fileName, fileNameToWriter, job);
  52.                 if (writer == null) {
  53.                     throw new IOException("Unable to create writer for path: " + fileName);
  54.                 }
  55.             }
  56.             writer.write(key, value);
  57.         }
  59.         @Override
  60.         public void close(TaskAttemptContext context) throws IOException, InterruptedException {
  61.             for (Entry<String, RecordWriter<K, V>> entry : fileNameToWriter.entrySet()) {
  62.                 entry.getValue().close(context);
  63.             }
  64.         }
  66.     }
  68.     private synchronized RecordWriter<K, V> createNewWriter(String folderName,
  69.             Map<String, RecordWriter<K, V>> fileNameToWriter, TaskAttemptContext job) {
  70.         try {
  71.             this.folderName = folderName;
  72.             RecordWriter<K, V> writer = super.getRecordWriter(job);
  73.             this.folderName = null;
  74.             fileNameToWriter.put(folderName, writer);
  75.             return writer;
  76.         } catch (Exception e) {
  77.             e.printStackTrace();
  78.             return null;
  79.         }
  80.     }
  82.     @Override
  83.     public Path getDefaultWorkFile(TaskAttemptContext context, String extension) throws IOException {
  84.         Path path = super.getDefaultWorkFile(context, extension);
  85.         if (folderName != null) {
  86.             String newPath = path.getParent().toString() + "/" + folderName + "/" + path.getName();
  87.             path = new Path(newPath);
  88.         }
  89.         return path;
  90.     }
  92.     @Override
  93.     public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
  94.         return new MultipleFilesRecordWriter(getFolderNameExtractor(), job);
  95.     }
  97.     public FolderNameExtractor<K, V> getFolderNameExtractor() {
  98.         return new KeyFolderNameExtractor<K, V>();
  99.     }
  101.     public interface FolderNameExtractor<K, V> {
  102.         public String extractFolderName(K key, V value);
  103.     }
  105.     private static class KeyFolderNameExtractor<K, V> implements FolderNameExtractor<K, V> {
  106.         public String extractFolderName(K key, V value) {
  107.             return key.toString();
  108.         }
  109.     }
  111. }
展开查看全部
赞(0）分享  回复(0）举报  2022-12-11
raogr8fs

raogr8fs3#

将根据减速器的数量生成多个输出文件。
您可以使用hadoop dfs -getmerge来合并输出。

赞(0）分享  回复(0）举报  2022-12-11
olqngx59

olqngx594#

您可以尝试更改输出文件名（Reducer输出），因为HDFS仅支持附加操作，那么它将从所有Reducer中收集所有Temp-r-0000 x文件（分区），并将它们放在一个文件中。
这里是您需要创建的类，它将覆盖TextOutputFormat中的方法：

  1. import java.io.IOException;
  2. import java.util.HashMap;
  3. import java.util.Map;
  4. import java.util.Map.Entry;
  6. import org.apache.hadoop.fs.Path;
  7. import org.apache.hadoop.mapreduce.RecordWriter;
  8. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  9. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  12. public class CustomNameMultipleFileOutputFormat<K, V> extends TextOutputFormat<K, V> {
  14.     private String folderName;
  16.     private class MultipleFilesRecordWriter extends RecordWriter<K, V> {
  18.         private Map<String, RecordWriter<K, V>> fileNameToWriter;
  19.         private FolderNameExtractor<K, V> fileNameExtractor;
  20.         private TaskAttemptContext job;
  21.         
  22.         
  24.         public MultipleFilesRecordWriter(FolderNameExtractor<K, V> fileNameExtractor, TaskAttemptContext job) {
  25.             fileNameToWriter = new HashMap<String, RecordWriter<K, V>>();
  26.             this.fileNameExtractor = fileNameExtractor;
  27.             this.job = job;
  28.         }
  30.         @Override
  31.         public void write(K key, V value) throws IOException, InterruptedException {
  32.             String fileName = "**[FOLDER_NAME_INCLUDING_SUB_DIRS]**";//fileNameExtractor.extractFolderName(key, value);
  33.             
  34.             RecordWriter<K, V> writer = fileNameToWriter.get(fileName);
  35.             if (writer == null) {
  36.                 writer = createNewWriter(fileName, fileNameToWriter, job);
  37.                 if (writer == null) {
  38.                     throw new IOException("Unable to create writer for path: " + fileName);
  39.                 }
  40.             }
  41.             writer.write(key, value);
  42.         }
  44.         @Override
  45.         public void close(TaskAttemptContext context) throws IOException, InterruptedException {
  46.             for (Entry<String, RecordWriter<K, V>> entry : fileNameToWriter.entrySet()) {
  47.                 entry.getValue().close(context);
  48.             }
  49.         }
  51.     
  53.     }
  55.     private synchronized RecordWriter<K, V> createNewWriter(String folderName,
  56.             Map<String, RecordWriter<K, V>> fileNameToWriter, TaskAttemptContext job) {
  57.         try {
  58.             this.folderName = folderName;
  59.             RecordWriter<K, V> writer = super.getRecordWriter(job);
  60.             this.folderName = null;
  61.             fileNameToWriter.put(folderName, writer);
  62.             
  63.             return writer;
  64.         } catch (Exception e) {
  65.             e.printStackTrace();
  66.             return null;
  67.         }
  68.     }
  70.     @Override
  71.     public Path getDefaultWorkFile(TaskAttemptContext context, String extension) throws IOException {
  72.         Path path = super.getDefaultWorkFile(context, extension);
  73.      
  74.         if (folderName != null) {
  75.             String newPath = path.getParent().toString() + "/" + folderName + "/**[ONE_FILE_NAME]**";
  76.             
  77.             path = new Path(newPath);
  78.       
  79.         }
  80.         return path;
  81.     }
  83.     @Override
  84.     public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
  85.         return new MultipleFilesRecordWriter(getFolderNameExtractor(), job);
  86.     }
  88.     public FolderNameExtractor<K, V> getFolderNameExtractor() {
  89.         return new KeyFolderNameExtractor<K, V>();
  90.     }
  92.     public interface FolderNameExtractor<K, V> {
  93.         public String extractFolderName(K key, V value);
  94.     }
  96.     private static class KeyFolderNameExtractor<K, V> implements FolderNameExtractor<K, V> {
  97.         public String extractFolderName(K key, V value) {
  98.             return key.toString();
  99.         }
  100.     }
  102. }

然后选择还原器/Map器：

  1. public static class ExtraLabReducer extends Reducer<CustomKeyComparable, Text, CustomKeyComparable, Text>
  2. {
  3.     MultipleOutputs multipleOutputs;
  5.     @Override
  6.     protected void setup(Context context) throws IOException, InterruptedException {
  7.         multipleOutputs = new MultipleOutputs(context);
  8.     }
  10.     @Override
  11.     public void reduce(CustomKeyComparable key, Iterable<Text> values, Context context) throws IOException, InterruptedException
  12.     {
  13.         for(Text d : values)
  14.         {
  15.              **multipleOutputs.write**("batta",key, d,**"[EXAMPLE_FILE_NAME]"**);
  16.         }
  17.         
  18.     }
  19.     
  20.     @Override
  21.     protected void cleanup(Context context) throws IOException, InterruptedException {
  22.         multipleOutputs.close();
  23.     }
  24.     
  25. }

然后在“作业配置：

  1. Job job = new Job(getConf(), "ExtraLab");
  2.     job.setJarByClass(ExtraLab.class);
  4.     job.setMapperClass(ExtraLabMapper.class);
  5.     job.setReducerClass(ExtraLabReducer.class);
  7.     job.setOutputKeyClass(Text.class);
  8.     job.setOutputValueClass(DoubleWritable.class);
  9.     
  10.     job.setMapOutputKeyClass(CustomKeyComparable.class);
  11.     job.setMapOutputValueClass(Text.class);
  13.     job.setInputFormatClass(TextInputFormat.class);
  14.     //job.setOutputFormatClass(TextOutputFormat.class);
  16.     
  17.     FileInputFormat.addInputPath(job, new Path(args[0]));
  18.     //adding one more reducer
  19.     job.setNumReduceTasks(2);
  20.     
  21.     LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
  23.     MultipleOutputs.addNamedOutput(job,"batta", CustomNameMultipleFileOutputFormat.class,CustomKeyComparable.class,Text.class);
展开查看全部
赞(0）分享  回复(0）举报  2022-12-11
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com