@Test public void testMultiOutputFormatWithReduce() throws Throwable { mrConf = mrCluster.createJobConf(); Job job = Job.getInstance(mrConf, "MultiOutWithReduce"); job.setMapperClass(WordCountMapper.class); job.setReducerClass(MultiOutWordCountReducer.class); job.setJarByClass(this.getClass()); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(MultiOutputFormat.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); JobConfigurer configurer = MultiOutputFormat.createConfigurer(job); configurer.addOutputFormat("out1", TextOutputFormat.class, IntWritable.class, Text.class); configurer.addOutputFormat("out2", SequenceFileOutputFormat.class, Text.class, IntWritable.class); configurer.addOutputFormat("out3", NullOutputFormat.class, Text.class, IntWritable.class); Path outDir = new Path(workDir.getPath(), job.getJobName()); FileOutputFormat.setOutputPath(configurer.getJob("out1"), new Path(outDir, "out1")); FileOutputFormat.setOutputPath(configurer.getJob("out2"), new Path(outDir, "out2")); configurer.configure(); String fileContent = "Hello World Hello World World"; String inputFile = createInputFile(fileContent); FileInputFormat.setInputPaths(job, new Path(inputFile)); Assert.assertTrue(job.waitForCompletion(true)); Path textOutPath = new Path(outDir, "out1/part-r-00000"); String[] textOutput = readFully(textOutPath).split("\n"); Path seqOutPath = new Path(outDir, "out2/part-r-00000"); SequenceFile.Reader reader = new SequenceFile.Reader(fs, seqOutPath, mrConf); Text key = new Text(); IntWritable value = new IntWritable(); String[] words = "Hello World".split(" "); Assert.assertEquals(words.length, textOutput.length); for (int i = 0; i < words.length; i++) { Assert.assertEquals((i + 2) + "\t" + words[i], textOutput[i]); reader.next(key, value); Assert.assertEquals(words[i], key.toString()); Assert.assertEquals((i + 2), value.get()); } Assert.assertFalse(reader.next(key, value)); }
public static JobConfigurer createConfigurer(Job job) { return JobConfigurer.create(job); }
MultiOutputFormat extends OutputFormat<Writable, Writable> { public static JobConfigurer createConfigurer(Job job) { return JobConfigurer.create(job); } }
MultiOutputFormat extends OutputFormat<Writable, Writable> { public static JobConfigurer createConfigurer(Job job) { return JobConfigurer.create(job); }  }
MultiOutputFormat extends OutputFormat<Writable, Writable> { public static JobConfigurer createConfigurer(Job job) { return JobConfigurer.create(job); }  static JobConfigurer createConfigurer(Job job); static JobContext getJobContext(String alias, JobContext context); static TaskAttemptContext getTaskAttemptContext(String alias, TaskAttemptContext context); static void write(String alias, K key, V value, TaskInputOutputContext context); @Override void checkOutputSpecs(JobContext context); @Override RecordWriter<Writable, Writable> getRecordWriter(TaskAttemptContext context); @Override OutputCommitter getOutputCommitter(TaskAttemptContext context); }
MultiOutputFormat extends OutputFormat<Writable, Writable> { public static JobConfigurer createConfigurer(Job job) { return JobConfigurer.create(job); }  static JobConfigurer createConfigurer(Job job); static JobContext getJobContext(String alias, JobContext context); static TaskAttemptContext getTaskAttemptContext(String alias, TaskAttemptContext context); static void write(String alias, K key, V value, TaskInputOutputContext context); @Override void checkOutputSpecs(JobContext context); @Override RecordWriter<Writable, Writable> getRecordWriter(TaskAttemptContext context); @Override OutputCommitter getOutputCommitter(TaskAttemptContext context);  }