CollocReducer extends MapReduceBase implements Reducer<GramKey,Gram,Gram,Gram> { @Override public void reduce(GramKey key, Iterator<Gram> values, OutputCollector<Gram,Gram> output, Reporter reporter) throws IOException { Gram.Type keyType = key.getType(); if (keyType == Gram.Type.UNIGRAM) { processUnigram(key, values, output, reporter); } else if (keyType == Gram.Type.HEAD || keyType == Gram.Type.TAIL) { processSubgram(key, values, output, reporter); } else { reporter.incrCounter(Skipped.MALFORMED_TYPES, 1); } }  @Override void configure(JobConf job); @Override void reduce(GramKey key,
                     Iterator<Gram> values,
                     OutputCollector<Gram,Gram> output,
                     Reporter reporter); static final String MIN_SUPPORT; static final int DEFAULT_MIN_SUPPORT; }
@Test public void testReduce() throws Exception { Gram[][] input = { {new Gram("the", UNIGRAM), new Gram("the", UNIGRAM), new Gram("the", UNIGRAM)}, {new Gram("the", HEAD), new Gram("the best", NGRAM), new Gram("the worst", NGRAM)}, {new Gram("of", HEAD), new Gram("of times", NGRAM), new Gram("of times", NGRAM)}, {new Gram("times", TAIL), new Gram("of times", NGRAM), new Gram("of times", NGRAM)}}; Gram[][] values = {{new Gram("the", 2, UNIGRAM), new Gram("the", 2, UNIGRAM)}, {new Gram("the best", 1, NGRAM), new Gram("the", 2, HEAD)}, {new Gram("the worst", 1, NGRAM), new Gram("the", 2, HEAD)}, {new Gram("of times", 2, NGRAM), new Gram("of", 2, HEAD)}, {new Gram("of times", 2, NGRAM), new Gram("times", 2, TAIL)}}; for (Gram[] v : values) { output.collect(v[0], v[1]); } EasyMock.replay(reporter, output); CollocReducer c = new CollocReducer(); GramKey key = new GramKey(); byte[] empty = new byte[0]; for (Gram[] ii : input) { key.set(ii[0], empty); List<Gram> vv = new LinkedList<Gram>(); vv.addAll(Arrays.asList(ii)); c.reduce(key, vv.iterator(), output, reporter); } EasyMock.verify(reporter, output); }