LLRReducer extends MapReduceBase implements Reducer<Gram,Gram,Text,DoubleWritable> { @Override public void reduce(Gram ngram, Iterator<Gram> values, OutputCollector<Text,DoubleWritable> output, Reporter reporter) throws IOException { int[] gramFreq = {-1, -1}; if (ngram.getType() == Gram.Type.UNIGRAM && emitUnigrams) { DoubleWritable dd = new DoubleWritable(ngram.getFrequency()); Text t = new Text(ngram.getString()); output.collect(t, dd); return; } String[] gram = new String[2]; while (values.hasNext()) { Gram value = values.next(); int pos = value.getType() == Gram.Type.HEAD ? 0 : 1; if (gramFreq[pos] != -1) { log.warn("Extra {} for {}, skipping", value.getType(), ngram); if (value.getType() == Gram.Type.HEAD) { reporter.incrCounter(Skipped.EXTRA_HEAD, 1); } else { reporter.incrCounter(Skipped.EXTRA_TAIL, 1); } return; } gram[pos] = value.getString(); gramFreq[pos] = value.getFrequency(); } if (gramFreq[0] == -1) { log.warn("Missing head for {}, skipping.", ngram); reporter.incrCounter(Skipped.MISSING_HEAD, 1); return; } else if (gramFreq[1] == -1) { log.warn("Missing tail for {}, skipping", ngram); reporter.incrCounter(Skipped.MISSING_TAIL, 1); return; } int k11 = ngram.getFrequency(); int k12 = gramFreq[0] - ngram.getFrequency(); int k21 = gramFreq[1] - ngram.getFrequency(); int k22 = (int) (ngramTotal - (gramFreq[0] + gramFreq[1] - ngram.getFrequency())); try { double llr = ll.logLikelihoodRatio(k11, k12, k21, k22); if (llr < minLLRValue) { reporter.incrCounter(Skipped.LESS_THAN_MIN_LLR, 1); return; } DoubleWritable dd = new DoubleWritable(llr); Text t = new Text(ngram.getString()); output.collect(t, dd); } catch (IllegalArgumentException ex) { reporter.incrCounter(Skipped.LLR_CALCULATION_ERROR, 1); log.error("Problem calculating LLR ratio: " + ex.getMessage()); log.error("NGram: " + ngram); log.error("HEAD: " + gram[0] + ':' + gramFreq[0]); log.error("TAIL: " + gram[1] + ':' + gramFreq[1]); log.error("k11: " + k11 + " k12: " + k12 + " k21: " + k21 + " k22: " + k22); } } LLRReducer();   LLRReducer(LLCallback ll); @Override void configure(JobConf job); @Override void reduce(Gram ngram,
                     Iterator<Gram> values,
                     OutputCollector<Text,DoubleWritable> output,
                     Reporter reporter); static final String NGRAM_TOTAL; static final String MIN_LLR; static final float DEFAULT_MIN_LLR; }
@Test public void testReduce() throws Exception { LLRReducer reducer = new LLRReducer(ll); Gram[][] input = { {new Gram("the best", 1, NGRAM), new Gram("the", 2, HEAD), new Gram("best", 1, TAIL) }, {new Gram("best of", 1, NGRAM), new Gram("best", 1, HEAD), new Gram("of", 2, TAIL) }, {new Gram("of times", 2, NGRAM), new Gram("of", 2, HEAD), new Gram("times", 2, TAIL) }, {new Gram("times the", 1, NGRAM), new Gram("times", 1, HEAD), new Gram("the", 1, TAIL) }, {new Gram("the worst", 1, NGRAM), new Gram("the", 2, HEAD), new Gram("worst", 1, TAIL) }, {new Gram("worst of", 1, NGRAM), new Gram("worst", 1, HEAD), new Gram("of", 2, TAIL) } }; int[][] expectations = { {1, 1, 0, 5}, {1, 0, 1, 5}, {2, 0, 0, 5}, {1, 0, 0, 6}, {1, 1, 0, 5}, {1, 0, 1, 5} }; for (int[] ee: expectations) { EasyMock.expect(ll.logLikelihoodRatio(ee[0], ee[1], ee[2], ee[3])).andDelegateTo(cl); } EasyMock.replay(ll); JobConf config = new JobConf(CollocDriver.class); config.set(LLRReducer.NGRAM_TOTAL, "7"); reducer.configure(config); for (Gram[] ii: input) { List<Gram> vv = new LinkedList<Gram>(); vv.addAll(Arrays.asList(ii).subList(1, ii.length)); reducer.reduce(ii[0], vv.iterator(), collector, reporter); } EasyMock.verify(ll); }