package com.smilezl.learn.CalWord;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class MultipleMapper {
public static class DeptMapper extends Mapper<Object, Text, Text, Text> {
@Override
protected void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
StringTokenizer tokenizer = new StringTokenizer(value.toString());
String tmp = "";
while (tokenizer.hasMoreTokens()) {
tmp += tokenizer.nextToken().toString();
}
context.write(new Text("dept"), new Text(tmp));
}
}
public static class EmpMapper extends Mapper<Object, Text, Text, Text> {
@Override
protected void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
StringTokenizer tokenizer = new StringTokenizer(value.toString());
String tmp = "";
while (tokenizer.hasMoreTokens()) {
tmp += tokenizer.nextToken().toString();
}
context.write(new Text("dept"), new Text(tmp));
}
}
public static class MulReducer extends Reducer<Object, Text, Text, Text> {
@Override
protected void reduce(Object key, Iterable<Text> value, Context context)
throws IOException, InterruptedException {
StringTokenizer tokenizer = new StringTokenizer(value.toString());
String tmp = "";
while (tokenizer.hasMoreTokens()) {
tmp += tokenizer.nextToken().toString() + "__";
}
context.write(new Text("emp"), new Text(tmp));
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 3) {
System.out.println(otherArgs.length);
System.out.println("Usage: MultipleMapper <in> <in> <out>");
System.exit(2);
}
Job job = new Job(conf, "MultipleMapper");
job.setJarByClass(MultipleMapper.class);
MultipleInputs.addInputPath(job, new Path(otherArgs[0]), TextInputFormat.class, DeptMapper.class);
MultipleInputs.addInputPath(job, new Path(otherArgs[1]), TextInputFormat.class, EmpMapper.class);
FileOutputFormat.setOutputPath(job, new Path(otherArgs[2]));
job.setCombinerClass(MulReducer.class);
job.setReducerClass(MulReducer.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
hadoop 多个maper处理,布布扣,bubuko.com
原文：http://smilezhuolin.blog.51cto.com/7671611/1430413