CollapserReducer extends Reducer<AvroKey<String>, AvroValue<IndexedRecord>, AvroKey<IndexedRecord>, NullWritable> { @Override public void reduce(AvroKey<String> key, Iterable<AvroValue<IndexedRecord>> values, Context context) throws IOException, InterruptedException { Iterator<AvroValue<IndexedRecord>> iterator = values.iterator(); List<IndexedRecord> objects = new ArrayList<IndexedRecord>(); while (iterator.hasNext()) { AvroValue<IndexedRecord> value = iterator.next(); objects.add((IndexedRecord) AvroUtils.getCopy(value.datum(), inputSchema, inputSchemaClass)); } List<IndexedRecord> collapsedList = recordCollapser.collapse(objects); for (IndexedRecord collapsed : collapsedList) { context.write(new AvroKey<IndexedRecord>(collapsed), NullWritable.get()); } }  @Override void reduce(AvroKey<String> key, Iterable<AvroValue<IndexedRecord>> values, Context context); static final String RECORD_COLLAPSER; static final String INPUT_SCHEMA; }
@Test public void testReduce() throws Exception { Configuration conf = new Configuration(); conf.set(CollapserReducer.RECORD_COLLAPSER, DummyDocumentToProjectCollapser.class.getCanonicalName()); conf.set(CollapserReducer.INPUT_SCHEMA, DocumentToProject.class.getCanonicalName()); doReturn(conf).when(context).getConfiguration(); reducer.setup(context); String docId1 = "docId1"; String docId2 = "docId2"; String projId1 = "projId1"; String projId2 = "projId2"; List<AvroValue<IndexedRecord>> docProjList = new ArrayList<>(); docProjList.add(buildInputRecord(docId1, projId1)); docProjList.add(buildInputRecord(docId1, projId2)); docProjList.add(buildInputRecord(docId2, projId1)); docProjList.add(buildInputRecord(docId2, projId2)); reducer.reduce(null, docProjList, context); verify(context, times(2)).write(keyCaptor.capture(), valueCaptor.capture()); assertEquals(2, keyCaptor.getAllValues().size()); assertTrue(keyCaptor.getAllValues().get(0).datum() instanceof Identifier); assertEquals(docId1, ((Identifier)keyCaptor.getAllValues().get(0).datum()).getId().toString()); assertTrue(keyCaptor.getAllValues().get(1).datum() instanceof Identifier); assertEquals(docId2, ((Identifier)keyCaptor.getAllValues().get(1).datum()).getId().toString()); assertEquals(2, valueCaptor.getAllValues().size()); assertTrue(NullWritable.get() == valueCaptor.getAllValues().get(0)); assertTrue(NullWritable.get() == valueCaptor.getAllValues().get(1)); }