title: MultipleOutputs Zero Reducer
body: <p>My input folder has 200 files. I want MultipleOutputs to write my parsed inputs from each file(identified using "map.input.file") into an output file with the same name.Since, I don't have any aggregation to perform and so using 0 reducer option (conf.setNumReduceTasks(0)). Ideally, I should get 200 output files.</p>
  
  <p>But, my output has  around 5000+ files - each file containing only one line (of streaming output).Clearly, it is not aggregating.My assumption is, Ideally, in zero reducer - mapper output should be aggregated . </p>
  
  <p>Help is appreciated. Thanks! </p>
  
  <pre><code> public static void main(String[] args) throws IOException {
  if (args.length != 2) {
    System.err.println("Usage: MaxTemperature &lt;input path&gt; &lt;output path&gt;");
    System.exit(-1);
  }    
  
  JobConf conf = new JobConf(MultipleOutputEx.class);
  conf.setJobName("Duration Count");
  
  FileInputFormat.addInputPath(conf, new Path(args[0]));
  FileOutputFormat.setOutputPath(conf, new Path(args[1]));
  
  conf.setNumReduceTasks(0);
  conf.setMapperClass(MultipleOutputExMapper.class);
  conf.setReducerClass(MultipleOutputExReducer.class);
  conf.setMapOutputKeyClass(NullWritable.class);            
  MultipleOutputs.addMultiNamedOutput(conf,"mofiles", TextOutputFormat.class, NullWritable.class, Text.class);    
  JobClient.runJob(conf);
  </code></pre>
  
  <p>}</p>
  
  <p>And my Mapper class is, </p>
  
  <pre><code>public class MultipleOutputExMapper extends MapReduceBase implements
      Mapper&lt;LongWritable, Text, NullWritable, Text&gt; {
  
  MultipleOutputs mos = null;
  Text fileKey = new Text();
  String line = "";
  private JobConf conf;
  
  @Override
  public void configure(JobConf conf) {
      this.conf = conf;
      mos = new MultipleOutputs(conf);
  }
  
  public void map(LongWritable key, Text value,
          OutputCollector&lt;NullWritable, Text&gt; output, Reporter reporter)
          throws IOException {
      try {
          String filename = conf.get("map.input.file");
          fileKey.set(filename);
          OutputCollector&lt;NullWritable, Text&gt; collector = mos.getCollector(
                  "mofiles", key.toString(), reporter);
          collector.collect(NullWritable.get(), value);
  
      } catch (ArrayIndexOutOfBoundsException E) {
          E.printStackTrace();
      } catch (Exception E) {
          System.out.println(line);
          E.printStackTrace();
      }
  }
  
  @Override
  public void close() throws IOException {
      mos.close();
  }
  </code></pre>