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); } }
