@Test public void testReduce() { TaggedKey key = new TaggedKey(1, 1); List<Text> values = new ArrayList<Text>(); values.add(new Text("Stephanie Leung,555-555-5555,123 No Such St,Fantasyland,CA 99999")); values.add(new Text("A,99.99,01/01/2013")); Text expectedValue = new Text( "Stephanie Leung,555-555-5555,123 No Such St,Fantasyland,CA 99999|A,99.99,01/01/2013"); reduceDriver.withReducer(reducer).withInput(key, values).withOutput(new IntWritable(1), expectedValue) .runTest(); }
@Override protected void reduce(TaggedKey key, Iterable<Text> values, Context ctx) throws IOException, InterruptedException { StringBuilder value = new StringBuilder(); LOGGER.debug("Entering reduce"); for (Text v : values) { LOGGER.debug("Loop - Key: {}, Value: {}.", key, value); value.append(v.toString()).append("|"); } value.delete(value.length() - 1, value.length()); LOGGER.debug("Key: {}, Value: {}.", key.getJoinKey(), value); ctx.write(key.getJoinKey(), new Text(value.toString())); }
ReduceSideJoinReducer extends Reducer<TaggedKey, Text, IntWritable, Text> { @Override protected void reduce(TaggedKey key, Iterable<Text> values, Context ctx) throws IOException, InterruptedException { StringBuilder value = new StringBuilder(); LOGGER.debug("Entering reduce"); for (Text v : values) { LOGGER.debug("Loop - Key: {}, Value: {}.", key, value); value.append(v.toString()).append("|"); } value.delete(value.length() - 1, value.length()); LOGGER.debug("Key: {}, Value: {}.", key.getJoinKey(), value); ctx.write(key.getJoinKey(), new Text(value.toString())); } }
ReduceSideJoinReducer extends Reducer<TaggedKey, Text, IntWritable, Text> { @Override protected void reduce(TaggedKey key, Iterable<Text> values, Context ctx) throws IOException, InterruptedException { StringBuilder value = new StringBuilder(); LOGGER.debug("Entering reduce"); for (Text v : values) { LOGGER.debug("Loop - Key: {}, Value: {}.", key, value); value.append(v.toString()).append("|"); } value.delete(value.length() - 1, value.length()); LOGGER.debug("Key: {}, Value: {}.", key.getJoinKey(), value); ctx.write(key.getJoinKey(), new Text(value.toString())); }  }
ReduceSideJoinReducer extends Reducer<TaggedKey, Text, IntWritable, Text> { @Override protected void reduce(TaggedKey key, Iterable<Text> values, Context ctx) throws IOException, InterruptedException { StringBuilder value = new StringBuilder(); LOGGER.debug("Entering reduce"); for (Text v : values) { LOGGER.debug("Loop - Key: {}, Value: {}.", key, value); value.append(v.toString()).append("|"); } value.delete(value.length() - 1, value.length()); LOGGER.debug("Key: {}, Value: {}.", key.getJoinKey(), value); ctx.write(key.getJoinKey(), new Text(value.toString())); }   }
ReduceSideJoinReducer extends Reducer<TaggedKey, Text, IntWritable, Text> { @Override protected void reduce(TaggedKey key, Iterable<Text> values, Context ctx) throws IOException, InterruptedException { StringBuilder value = new StringBuilder(); LOGGER.debug("Entering reduce"); for (Text v : values) { LOGGER.debug("Loop - Key: {}, Value: {}.", key, value); value.append(v.toString()).append("|"); } value.delete(value.length() - 1, value.length()); LOGGER.debug("Key: {}, Value: {}.", key.getJoinKey(), value); ctx.write(key.getJoinKey(), new Text(value.toString())); }   static final Logger LOGGER; }