DuplicateWorkDetectReduceService implements DiReduceService<Text, BytesWritable, Text, Text> { void process(Text key, Reducer<Text, BytesWritable, Text, Text>.Context context, Iterable<BytesWritable> documents, int maxNumberOfDocuments) throws IOException, InterruptedException { Iterator<BytesWritable> docsIterator = documents.iterator(); log.info("-- start process, key: {}", key.toString()); int partNb = 1; while (docsIterator.hasNext()) { int counter = 0; List<DocumentProtos.DocumentMetadata> metadataList = new ArrayList<DocumentProtos.DocumentMetadata>(); while (counter < maxNumberOfDocuments && docsIterator.hasNext()) { DocumentProtos.DocumentWrapper docWrapper = DocumentProtos.DocumentWrapper.parseFrom(docsIterator.next().copyBytes()); metadataList.add(docWrapper.getDocumentMetadata()); counter++; } if (counter > 0) { log.info("---- key " + key.toString() + ", part "+ partNb + ", documents {} " + counter); if (isDebugMode(context.getConfiguration())) { StringBuilder debugInfo = new StringBuilder(); duplicateWorkService.findDuplicates(metadataList, debugInfo); context.write(key, new Text(debugInfo.toString())); } else { Map<Integer, Set<DocumentProtos.DocumentMetadata>> duplicateWorksMap = duplicateWorkService.findDuplicates(metadataList, null); saveDuplicatesToContext(duplicateWorksMap, key, context); } context.progress(); } partNb++; } log.info("-- end process, key: {}", key); }  @Override void reduce(Text key, Iterable<BytesWritable> values, Reducer<Text, BytesWritable, Text, Text>.Context context); @Value("1000") void setBeginPackSize(int beginPackSize);  }
@Test public void testProcess() throws Exception { long startTime = new Date().getTime(); duplicateWorkDetectReduceService.process(new Text(""), context, documentsBW, 100); long endTime = new Date().getTime(); log.info("time[ms]: " + (endTime-startTime)); }