Java 类org.apache.hadoop.mapreduce.lib.chain.ChainMapper 实例源码

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();

    Job job = Job.getInstance(conf);
    job.setJarByClass(Multiplication.class);

    ChainMapper.addMapper(job, CooccurrenceMapper.class, LongWritable.class, Text.class, Text.class, Text.class, conf);
    ChainMapper.addMapper(job, RatingMapper.class, Text.class, Text.class, Text.class, Text.class, conf);

    job.setMapperClass(CooccurrenceMapper.class);
    job.setMapperClass(RatingMapper.class);

    job.setReducerClass(MultiplicationReducer.class);

    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(DoubleWritable.class);

    MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, CooccurrenceMapper.class);
    MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, RatingMapper.class);

    TextOutputFormat.setOutputPath(job, new Path(args[2]));

    job.waitForCompletion(true);
}

public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(UnitMultiplication.class);

        ChainMapper.addMapper(job, TransitionMapper.class, Object.class, Text.class, Text.class, Text.class, conf);
        ChainMapper.addMapper(job, PRMapper.class, Object.class, Text.class, Text.class, Text.class, conf);

        job.setReducerClass(MultiplicationReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, TransitionMapper.class);
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, PRMapper.class);

        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        job.waitForCompletion(true);
    }

public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setFloat("beta", Float.parseFloat(args[3]));
        Job job = Job.getInstance(conf);
        job.setJarByClass(UnitMultiplication.class);

        ChainMapper.addMapper(job, TransitionMapper.class, Object.class, Text.class, Text.class, Text.class, conf);
        ChainMapper.addMapper(job, PRMapper.class, Object.class, Text.class, Text.class, Text.class, conf);

        job.setReducerClass(MultiplicationReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, TransitionMapper.class);
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, PRMapper.class);

        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        job.waitForCompletion(true);
    }

public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setFloat("beta", Float.parseFloat(args[3]));
        Job job = Job.getInstance(conf);
        job.setJarByClass(UnitSum.class);

        ChainMapper.addMapper(job, PassMapper.class, Object.class, Text.class, Text.class, DoubleWritable.class, conf);
        ChainMapper.addMapper(job, BetaMapper.class, Text.class, DoubleWritable.class, Text.class, DoubleWritable.class, conf);

        job.setReducerClass(SumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, PassMapper.class);
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, BetaMapper.class);

        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        job.waitForCompletion(true);
    }

@Override
public void addMapReduce(final Class<? extends Mapper> mapper,
                         final Class<? extends Reducer> combiner,
                         final Class<? extends Reducer> reducer,
                         final Class<? extends WritableComparator> comparator,
                         final Class<? extends WritableComparable> mapOutputKey,
                         final Class<? extends WritableComparable> mapOutputValue,
                         final Class<? extends WritableComparable> reduceOutputKey,
                         final Class<? extends WritableComparable> reduceOutputValue,
                         final Configuration configuration) {

   Configuration mergedConf = overlayConfiguration(getConf(), configuration);

   try {
        final Job job;

        if (State.NONE == this.state || State.REDUCER == this.state) {
            // Create a new job with a reference to mergedConf
            job = Job.getInstance(mergedConf);
            job.setJobName(makeClassName(mapper) + ARROW + makeClassName(reducer));
            HBaseAuthHelper.setHBaseAuthToken(mergedConf, job);
            this.jobs.add(job);
        } else {
            job = this.jobs.get(this.jobs.size() - 1);
            job.setJobName(job.getJobName() + ARROW + makeClassName(mapper) + ARROW + makeClassName(reducer));
        }
        job.setNumReduceTasks(this.getConf().getInt("mapreduce.job.reduces", this.getConf().getInt("mapreduce.tasktracker.reduce.tasks.maximum", 1)));

        ChainMapper.addMapper(job, mapper, NullWritable.class, FaunusVertex.class, mapOutputKey, mapOutputValue, mergedConf);
        ChainReducer.setReducer(job, reducer, mapOutputKey, mapOutputValue, reduceOutputKey, reduceOutputValue, mergedConf);

        if (null != comparator)
            job.setSortComparatorClass(comparator);
        if (null != combiner)
            job.setCombinerClass(combiner);
        if (null == job.getConfiguration().get(MAPREDUCE_MAP_OUTPUT_COMPRESS, null))
            job.getConfiguration().setBoolean(MAPREDUCE_MAP_OUTPUT_COMPRESS, true);
        if (null == job.getConfiguration().get(MAPREDUCE_MAP_OUTPUT_COMPRESS_CODEC, null))
            job.getConfiguration().setClass(MAPREDUCE_MAP_OUTPUT_COMPRESS_CODEC, DefaultCodec.class, CompressionCodec.class);
        this.state = State.REDUCER;
    } catch (IOException e) {
        throw new RuntimeException(e.getMessage(), e);
    }

}

@Override
public void addMap(final Class<? extends Mapper> mapper,
                   final Class<? extends WritableComparable> mapOutputKey,
                   final Class<? extends WritableComparable> mapOutputValue,
                   Configuration configuration) {

    Configuration mergedConf = overlayConfiguration(getConf(), configuration);

    try {
        final Job job;
        if (State.NONE == this.state) {
            // Create a new job with a reference to mergedConf
            job = Job.getInstance(mergedConf);
            job.setNumReduceTasks(0);
            job.setJobName(makeClassName(mapper));
            HBaseAuthHelper.setHBaseAuthToken(mergedConf, job);
            this.jobs.add(job);
        } else {
            job = this.jobs.get(this.jobs.size() - 1);
            job.setJobName(job.getJobName() + ARROW + makeClassName(mapper));
        }

        if (State.MAPPER == this.state || State.NONE == this.state) {
            ChainMapper.addMapper(job, mapper, NullWritable.class, FaunusVertex.class, mapOutputKey, mapOutputValue, mergedConf);
            /* In case no reducer is defined later for this job, set the job
             * output k/v to match the mapper output k-v.  Output formats that
             * care about their configured k-v classes (such as
             * SequenceFileOutputFormat) require these to be set correctly lest
             * they throw an exception at runtime.
             *
             * ChainReducer.setReducer overwrites these k-v settings, so if a
             * reducer is added onto this job later, these settings will be
             * overridden by the actual reducer's output k-v.
             */
            job.setOutputKeyClass(mapOutputKey);
            job.setOutputValueClass(mapOutputValue);
            this.state = State.MAPPER;
            logger.info("Added mapper " + job.getJobName() + " via ChainMapper with output (" + mapOutputKey + "," + mapOutputValue + "); current state is " + state);
        } else {
            ChainReducer.addMapper(job, mapper, NullWritable.class, FaunusVertex.class, mapOutputKey, mapOutputValue, mergedConf);
            this.state = State.REDUCER;
            logger.info("Added mapper " + job.getJobName() + " via ChainReducer with output (" + mapOutputKey + "," + mapOutputValue + "); current state is " + state);
        }
    } catch (IOException e) {
        throw new RuntimeException(e.getMessage(), e);
    }
}