@Test public void testGetContext() throws Exception { String namedOutputMeta = "meta"; String namedOutputFault = "fault"; Configuration conf = new Configuration(); conf.set(MULTIPLE_OUTPUTS, namedOutputMeta + ' ' + namedOutputFault); conf.set(MO_PREFIX + namedOutputMeta + ".keyschema", Identifier.SCHEMA$.toString()); conf.setClass(MO_PREFIX + namedOutputMeta + FORMAT, MockOutputFormat.class, OutputFormat.class); doReturn(MockOutputFormat.class).when(context).getOutputFormatClass(); doReturn(conf).when(context).getConfiguration(); doReturn(taskAttemptId).when(context).getTaskAttemptID(); doReturn(new JobID()).when(taskAttemptId).getJobID(); AvroMultipleOutputs multipleOutputs = new AvroMultipleOutputs(context); TaskAttemptContext attemptContext = multipleOutputs.getContext(namedOutputMeta); assertNotNull(attemptContext); assertEquals(Identifier.SCHEMA$.toString(), attemptContext.getConfiguration().get(MO_PREFIX + namedOutputMeta + ".keyschema")); }
@SuppressWarnings("deprecation") public TaskAttemptContext getContext(String nameOutput) throws IOException { TaskAttemptContext taskContext = taskContexts.get(nameOutput); if (taskContext != null) { return taskContext; } Job job = new Job(context.getConfiguration()); job.setOutputFormatClass(getNamedOutputFormatClass(context, nameOutput)); Schema keySchema=null,valSchema=null; if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".keyschema",null) != null) keySchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".keyschema")); if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".valueschema", null) != null) valSchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".valueschema")); setSchema(job, keySchema, valSchema); taskContext = createTaskAttemptContext( job.getConfiguration(), context.getTaskAttemptID()); taskContexts.put(nameOutput, taskContext); return taskContext; }
AvroMultipleOutputs { @SuppressWarnings("deprecation") public TaskAttemptContext getContext(String nameOutput) throws IOException { TaskAttemptContext taskContext = taskContexts.get(nameOutput); if (taskContext != null) { return taskContext; } Job job = new Job(context.getConfiguration()); job.setOutputFormatClass(getNamedOutputFormatClass(context, nameOutput)); Schema keySchema=null,valSchema=null; if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".keyschema",null) != null) keySchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".keyschema")); if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".valueschema", null) != null) valSchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".valueschema")); setSchema(job, keySchema, valSchema); taskContext = createTaskAttemptContext( job.getConfiguration(), context.getTaskAttemptID()); taskContexts.put(nameOutput, taskContext); return taskContext; } }
AvroMultipleOutputs { @SuppressWarnings("deprecation") public TaskAttemptContext getContext(String nameOutput) throws IOException { TaskAttemptContext taskContext = taskContexts.get(nameOutput); if (taskContext != null) { return taskContext; } Job job = new Job(context.getConfiguration()); job.setOutputFormatClass(getNamedOutputFormatClass(context, nameOutput)); Schema keySchema=null,valSchema=null; if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".keyschema",null) != null) keySchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".keyschema")); if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".valueschema", null) != null) valSchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".valueschema")); setSchema(job, keySchema, valSchema); taskContext = createTaskAttemptContext( job.getConfiguration(), context.getTaskAttemptID()); taskContexts.put(nameOutput, taskContext); return taskContext; } AvroMultipleOutputs(
      TaskInputOutputContext<?, ?, ?, ?> context); }
AvroMultipleOutputs { @SuppressWarnings("deprecation") public TaskAttemptContext getContext(String nameOutput) throws IOException { TaskAttemptContext taskContext = taskContexts.get(nameOutput); if (taskContext != null) { return taskContext; } Job job = new Job(context.getConfiguration()); job.setOutputFormatClass(getNamedOutputFormatClass(context, nameOutput)); Schema keySchema=null,valSchema=null; if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".keyschema",null) != null) keySchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".keyschema")); if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".valueschema", null) != null) valSchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".valueschema")); setSchema(job, keySchema, valSchema); taskContext = createTaskAttemptContext( job.getConfiguration(), context.getTaskAttemptID()); taskContexts.put(nameOutput, taskContext); return taskContext; } AvroMultipleOutputs(
      TaskInputOutputContext<?, ?, ?, ?> context); @SuppressWarnings("rawtypes") static void addNamedOutput(Job job, String namedOutput,
      Class<? extends OutputFormat> outputFormatClass,
      Schema keySchema); @SuppressWarnings("rawtypes") static void addNamedOutput(Job job, String namedOutput,
      Class<? extends OutputFormat> outputFormatClass,
      Schema keySchema, Schema valueSchema); static void setCountersEnabled(Job job, boolean enabled); static boolean getCountersEnabled(JobContext job); Set<String> getNamedOutputs(); void write(String namedOutput, Object key); void write(String namedOutput, Object key, Object value); @SuppressWarnings("unchecked") void write(String namedOutput, Object key, Object value,
      String baseOutputPath); void write(Object key, Object value, String baseOutputPath); @SuppressWarnings({ "unchecked", "deprecation" }) void write(Object key, Object value, Schema keySchema,
      Schema valSchema, String baseOutputPath); @SuppressWarnings("rawtypes") synchronized RecordWriter getRecordWriter(
      TaskAttemptContext taskContext, String baseFileName); @SuppressWarnings("deprecation") TaskAttemptContext getContext(String nameOutput); void close(); }
AvroMultipleOutputs { @SuppressWarnings("deprecation") public TaskAttemptContext getContext(String nameOutput) throws IOException { TaskAttemptContext taskContext = taskContexts.get(nameOutput); if (taskContext != null) { return taskContext; } Job job = new Job(context.getConfiguration()); job.setOutputFormatClass(getNamedOutputFormatClass(context, nameOutput)); Schema keySchema=null,valSchema=null; if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".keyschema",null) != null) keySchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".keyschema")); if (job.getConfiguration().get(MO_PREFIX + nameOutput + ".valueschema", null) != null) valSchema = Schema.parse(job.getConfiguration().get( MO_PREFIX + nameOutput + ".valueschema")); setSchema(job, keySchema, valSchema); taskContext = createTaskAttemptContext( job.getConfiguration(), context.getTaskAttemptID()); taskContexts.put(nameOutput, taskContext); return taskContext; } AvroMultipleOutputs(
      TaskInputOutputContext<?, ?, ?, ?> context); @SuppressWarnings("rawtypes") static void addNamedOutput(Job job, String namedOutput,
      Class<? extends OutputFormat> outputFormatClass,
      Schema keySchema); @SuppressWarnings("rawtypes") static void addNamedOutput(Job job, String namedOutput,
      Class<? extends OutputFormat> outputFormatClass,
      Schema keySchema, Schema valueSchema); static void setCountersEnabled(Job job, boolean enabled); static boolean getCountersEnabled(JobContext job); Set<String> getNamedOutputs(); void write(String namedOutput, Object key); void write(String namedOutput, Object key, Object value); @SuppressWarnings("unchecked") void write(String namedOutput, Object key, Object value,
      String baseOutputPath); void write(Object key, Object value, String baseOutputPath); @SuppressWarnings({ "unchecked", "deprecation" }) void write(Object key, Object value, Schema keySchema,
      Schema valSchema, String baseOutputPath); @SuppressWarnings("rawtypes") synchronized RecordWriter getRecordWriter(
      TaskAttemptContext taskContext, String baseFileName); @SuppressWarnings("deprecation") TaskAttemptContext getContext(String nameOutput); void close();  }