@Test public void testMultiOutputFormatWithoutReduce() throws Throwable { mrConf = mrCluster.createJobConf(); Job job = Job.getInstance(mrConf, "MultiOutNoReduce"); job.setMapperClass(MultiOutWordIndexMapper.class); job.setJarByClass(this.getClass()); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(MultiOutputFormat.class); job.setNumReduceTasks(0); JobConfigurer configurer = MultiOutputFormat.createConfigurer(job); configurer.addOutputFormat("out1", TextOutputFormat.class, IntWritable.class, Text.class); configurer.addOutputFormat("out2", SequenceFileOutputFormat.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")); String fileContent = "Hello World"; String inputFile = createInputFile(fileContent); FileInputFormat.setInputPaths(job, new Path(inputFile)); DistributedCache.addFileToClassPath(new Path(inputFile), job.getConfiguration(), fs); String dummyFile = createInputFile("dummy file"); DistributedCache.addFileToClassPath(new Path(dummyFile), configurer.getJob("out1") .getConfiguration(), fs); DistributedCache.addFileToClassPath(new Path(inputFile), configurer.getJob("out2") .getConfiguration(), fs); configurer.configure(); Path[] fileClassPaths = DistributedCache.getFileClassPaths(job.getConfiguration()); List<Path> fileClassPathsList = Arrays.asList(fileClassPaths); Assert.assertTrue("Cannot find " + (new Path(inputFile)) + " in " + fileClassPathsList, fileClassPathsList.contains(new Path(inputFile))); Assert.assertTrue("Cannot find " + (new Path(dummyFile)) + " in " + fileClassPathsList, fileClassPathsList.contains(new Path(dummyFile))); URI[] cacheFiles = DistributedCache.getCacheFiles(job.getConfiguration()); List<URI> cacheFilesList = Arrays.asList(cacheFiles); URI inputFileURI = new Path(inputFile).makeQualified(fs).toUri(); Assert.assertTrue("Cannot find " + inputFileURI + " in " + cacheFilesList, cacheFilesList.contains(inputFileURI)); URI dummyFileURI = new Path(dummyFile).makeQualified(fs).toUri(); Assert.assertTrue("Cannot find " + dummyFileURI + " in " + cacheFilesList, cacheFilesList.contains(dummyFileURI)); Assert.assertTrue(job.waitForCompletion(true)); Path textOutPath = new Path(outDir, "out1/part-m-00000"); String[] textOutput = readFully(textOutPath).split("\n"); Path seqOutPath = new Path(outDir, "out2/part-m-00000"); SequenceFile.Reader reader = new SequenceFile.Reader(fs, seqOutPath, mrConf); Text key = new Text(); IntWritable value = new IntWritable(); String[] words = fileContent.split(" "); Assert.assertEquals(words.length, textOutput.length); LOG.info("Verifying file contents"); for (int i = 0; i < words.length; i++) { Assert.assertEquals((i + 1) + "\t" + words[i], textOutput[i]); reader.next(key, value); Assert.assertEquals(words[i], key.toString()); Assert.assertEquals((i + 1), 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);  }