import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
@Override
public void process(Annotation annotation, Job job, Object target)
		throws ToolException {
	AvroNamedOutput avroOut = (AvroNamedOutput)annotation;

	Schema schema = getSchema(avroOut.record());
	String[] names = getNames(avroOut);
	for (String name : names) {
		name = (String)evaluateExpression(name);
		if (!configured.contains(name)) {
			AvroMultipleOutputs.addNamedOutput(job, name, avroOut.format(), schema);
			AvroMultipleOutputs.setCountersEnabled(job, avroOut.countersEnabled());
			configured.add(name);
		}
	}

	AvroSerialization.addToConfiguration(job.getConfiguration());
}
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
@Override
public void setContext(TaskInputOutputContext<Object,Object,Object,Object> context)
{           
  super.setContext(context);
  
  // ... and we also write the final output to multiple directories
  _multipleOutputs = new AvroMultipleOutputs(context);
}
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
public Job internalRun(Path origInput, Path destInput, Path outputDir, Configuration conf) throws Exception {

        conf.set("viadeo.diff.diffinpath", origInput.toString());
        conf.set("viadeo.diff.diffoutpath", destInput.toString());

        Job job = new Job(conf);
        job.setJarByClass(DiffJob.class);
        job.setJobName("diff");

        Schema schema = SchemaUtils.getConfSchema(conf);
        if(schema == null) schema = SchemaUtils.getSchema(conf, destInput);


        FileInputFormat.setInputPaths(job, origInput, destInput);
        job.setInputFormatClass(AvroKeyInputFormat.class);

        job.setMapperClass(DiffMapper.class);
        AvroJob.setInputKeySchema(job, schema);
        AvroJob.setMapOutputKeySchema(job, schema);
        job.setMapOutputValueClass(Text.class);
        job.setReducerClass(DiffReducer.class);
        AvroJob.setOutputKeySchema(job, schema);

        job.setOutputValueClass(Text.class);
        job.setOutputFormatClass(AvroKeyOutputFormat.class);


        // ~ OUTPUT
        FileOutputFormat.setOutputPath(job, outputDir);
        AvroMultipleOutputs.addNamedOutput(job, "kernel", AvroKeyOutputFormat.class, schema);
        AvroMultipleOutputs.addNamedOutput(job, "add", AvroKeyOutputFormat.class, schema);
        AvroMultipleOutputs.addNamedOutput(job, "del", AvroKeyOutputFormat.class, schema);

        AvroMultipleOutputs.setCountersEnabled(job, true);

        return job;
    }
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
@Override
public void setup(Context context) throws IOException, InterruptedException {
  LOGGER.info("DerivedColumnTransformationPhaseJob.DerivedColumnTransformationPhaseMapper.setup()");
  Configuration configuration = context.getConfiguration();
  FileSystem fs = FileSystem.get(configuration);

  FileSplit fileSplit = (FileSplit) context.getInputSplit();
  inputFileName = fileSplit.getPath().getName();
  inputFileName = inputFileName.substring(0, inputFileName.lastIndexOf(ThirdEyeConstants.AVRO_SUFFIX));
  LOGGER.info("split name:" + inputFileName);

  thirdeyeConfig = OBJECT_MAPPER.readValue(configuration.get(DERIVED_COLUMN_TRANSFORMATION_PHASE_THIRDEYE_CONFIG.toString()), ThirdEyeConfig.class);
  config = DerivedColumnTransformationPhaseConfig.fromThirdEyeConfig(thirdeyeConfig);
  dimensionsNames = config.getDimensionNames();
  dimensionsTypes = config.getDimensionTypes();
  metricNames = config.getMetricNames();
  metricTypes = config.getMetricTypes();
  timeColumnName = config.getTimeColumnName();
  whitelist = config.getWhitelist();
  nonWhitelistValueMap = config.getNonWhitelistValue();

  outputSchema = new Schema.Parser().parse(configuration.get(DERIVED_COLUMN_TRANSFORMATION_PHASE_OUTPUT_SCHEMA.toString()));

  Path topKPath = new Path(configuration.get(DERIVED_COLUMN_TRANSFORMATION_PHASE_TOPK_PATH.toString())
      + File.separator + ThirdEyeConstants.TOPK_VALUES_FILE);
  topKDimensionValues = new TopKDimensionValues();
  if (fs.exists(topKPath)) {
    FSDataInputStream topkValuesStream = fs.open(topKPath);
    topKDimensionValues = OBJECT_MAPPER.readValue((DataInput) topkValuesStream, TopKDimensionValues.class);
    topkValuesStream.close();
  }
  topKDimensionsMap = topKDimensionValues.getTopKDimensions();

  avroMultipleOutputs = new AvroMultipleOutputs(context);
}
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
protected AvroMultipleOutputs getDefaultAvroNamedOutput() {
	return getAvroNamedOutput("default");
}
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
protected AvroMultipleOutputs getAvroNamedOutput(String name) {
	return this.avroNamedOutputs.get(name);
}
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
@Override
protected void setup(org.apache.hadoop.mapreduce.Reducer<RecordKey, AvroValue<Record>, NullWritable, AvroValue<Record>>.Context context)
  throws IOException, InterruptedException {
  multipleOutputsAvro = new AvroMultipleOutputs(context);
  partitions.clear();
}
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
@Override
protected void setup(Context context) {
    amos = new AvroMultipleOutputs(context);
}
 

import org.apache.avro.mapreduce.AvroMultipleOutputs; //导入依赖的package包/类
public Job run() throws Exception {
  Job job = Job.getInstance(getConf());
  job.setJobName(name);
  job.setJarByClass(DerivedColumnTransformationPhaseJob.class);

  Configuration configuration = job.getConfiguration();
  FileSystem fs = FileSystem.get(configuration);

  // Input Path
  String inputPathDir = getAndSetConfiguration(configuration, DERIVED_COLUMN_TRANSFORMATION_PHASE_INPUT_PATH);
  LOGGER.info("Input path dir: " + inputPathDir);
  for (String inputPath : inputPathDir.split(",")) {
    LOGGER.info("Adding input:" + inputPath);
    Path input = new Path(inputPath);
    FileInputFormat.addInputPath(job, input);
  }

  // Topk path
  String topkPath = getAndSetConfiguration(configuration, DERIVED_COLUMN_TRANSFORMATION_PHASE_TOPK_PATH);
  LOGGER.info("Topk path : " + topkPath);

  // Output path
  Path outputPath = new Path(getAndSetConfiguration(configuration, DERIVED_COLUMN_TRANSFORMATION_PHASE_OUTPUT_PATH));
  LOGGER.info("Output path dir: " + outputPath.toString());
  if (fs.exists(outputPath)) {
    fs.delete(outputPath, true);
  }
  FileOutputFormat.setOutputPath(job, outputPath);

  // Schema
  Schema avroSchema = ThirdeyeAvroUtils.getSchema(inputPathDir);
  LOGGER.info("Schema : {}", avroSchema.toString(true));

  // ThirdEyeConfig
  String dimensionTypesProperty = ThirdeyeAvroUtils.getDimensionTypesProperty(
      props.getProperty(ThirdEyeConfigProperties.THIRDEYE_DIMENSION_NAMES.toString()), avroSchema);
  props.setProperty(ThirdEyeConfigProperties.THIRDEYE_DIMENSION_TYPES.toString(), dimensionTypesProperty);
  String metricTypesProperty = ThirdeyeAvroUtils.getMetricTypesProperty(
      props.getProperty(ThirdEyeConfigProperties.THIRDEYE_METRIC_NAMES.toString()),
      props.getProperty(ThirdEyeConfigProperties.THIRDEYE_METRIC_TYPES.toString()), avroSchema);
  props.setProperty(ThirdEyeConfigProperties.THIRDEYE_METRIC_TYPES.toString(), metricTypesProperty);
  ThirdEyeConfig thirdeyeConfig = ThirdEyeConfig.fromProperties(props);
  job.getConfiguration().set(DERIVED_COLUMN_TRANSFORMATION_PHASE_THIRDEYE_CONFIG.toString(),
      OBJECT_MAPPER.writeValueAsString(thirdeyeConfig));
  LOGGER.info("ThirdEyeConfig {}", thirdeyeConfig.encode());

  // New schema
  Schema outputSchema = newSchema(thirdeyeConfig);
  job.getConfiguration().set(DERIVED_COLUMN_TRANSFORMATION_PHASE_OUTPUT_SCHEMA.toString(), outputSchema.toString());

  // Map config
  job.setMapperClass(DerivedColumnTransformationPhaseMapper.class);
  job.setInputFormatClass(AvroKeyInputFormat.class);
  job.setMapOutputKeyClass(AvroKey.class);
  job.setMapOutputValueClass(NullWritable.class);
  AvroJob.setOutputKeySchema(job, outputSchema);
  LazyOutputFormat.setOutputFormatClass(job, AvroKeyOutputFormat.class);
  AvroMultipleOutputs.addNamedOutput(job, "avro", AvroKeyOutputFormat.class, outputSchema);

  job.setNumReduceTasks(0);

  job.waitForCompletion(true);

  return job;
}