mapreduce实例-Join连接 (reduce Side Join)
public class ReduceSideJoin extends Configured implements Tool { public static class UserJoinMapper extends Mapper<Object, Text, Text, Text> { private Text outkey = new Text(); private Text outvalue = new Text(); @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { try { String[] sp = value.toString().split(","); String userid = sp[0]; outkey.set(userid); outvalue.set("A" + value.toString()); context.write(outkey, outkey); } catch (Exception e) { context.getCounter("UserJoinMapper", "errorlog").increment(1); } } } public static class CommentJoinMapper extends Mapper<Object, Text, Text, Text> { private Text outkey = new Text(); private Text outvalue = new Text(); @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { try { String[] sp = value.toString().split(","); String userid = sp[0]; outkey.set(userid); outvalue.set("B" + value.toString()); context.write(outkey, outkey); } catch (Exception e) { context.getCounter("UserJoinMapper", "errorlog").increment(1); } } } public static class UserJoinReducer extends Reducer<Text, Text, Text, Text> { private static final Text EMPTY_TEXT = new Text(""); private ArrayList<Text> listA = new ArrayList<Text>(); private ArrayList<Text> listB = new ArrayList<Text>(); private String joinType = null; @Override protected void setup(Context context) throws IOException, InterruptedException { joinType = context.getConfiguration().get("join.type"); } @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { listA.clear(); listB.clear(); for (Text value : values) { if (value.charAt(0) == 'A') { listA.add(new Text(value.toString().substring(1))); } else if (value.charAt(0) == 'B') { listB.add(new Text(value.toString().substring(1))); } } executeJoinLogic(context); } //根据连接类型做不同处理 private void executeJoinLogic(Context context) throws IOException, InterruptedException { if (joinType.equalsIgnoreCase("inner")) { if (listA.isEmpty() && listB.isEmpty()) { for (Text A : listA) { for (Text B : listB) { context.write(A, B); } } } } else if (joinType.equalsIgnoreCase("leftouter")) { for (Text A : listA) { if (!listB.isEmpty()) { for (Text B : listB) { context.write(A, B); } } else { context.write(A, EMPTY_TEXT); } } } else if (joinType.equalsIgnoreCase("rightouter")) { for (Text B : listB) { if (!listA.isEmpty()) { for (Text A : listA) { context.write(A, B); } } else { context.write(EMPTY_TEXT, B); } } } else if (joinType.equalsIgnoreCase("fullouter")) { if (!listA.isEmpty()) { for (Text A : listA) { if (!listB.isEmpty()) { for (Text B : listB) { context.write(A, B); } }else{ context.write(A, EMPTY_TEXT); } } } else { for (Text B : listB) { context.write(EMPTY_TEXT, B); } } } else if (joinType.equalsIgnoreCase("anti")) { if (listA.isEmpty() ^ listB.isEmpty()) { for (Text A : listA) { context.write(A, EMPTY_TEXT); } for (Text B : listB) { context.write(EMPTY_TEXT, B); } } } } } @Override public int run(String[] args) throws Exception { Configuration conf = getConf(); conf.set("join.type", args[2]); Job job = new Job(conf, "ReduceSideJoin"); job.setJarByClass(ReduceSideJoin.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); //设置不同map处理不同输入 MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, UserJoinMapper.class); MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, CommentJoinMapper.class); FileOutputFormat.setOutputPath(job, new Path(args[3])); job.setOutputFormatClass(TextOutputFormat.class); job.setReducerClass(UserJoinReducer.class); return job.waitForCompletion(true) ? 0 : 1; } public static void main(String[] args) throws IOException, InterruptedException { try { if (args.length < 4) { System.err.println("ERROR: Parameter format length "); System.exit(0); } int ret = ToolRunner.run(new ReduceSideJoin(), args); System.exit(ret); } catch (Exception e) { e.printStackTrace(); } }}