@Test public void testReduce() throws Exception { LLRReducer reducer = new LLRReducer(ll); Gram[][] input = { {new Gram("the best", 1, Gram.Type.NGRAM), new Gram("the", 2, Gram.Type.HEAD), new Gram("best", 1, Gram.Type.TAIL) }, {new Gram("best of", 1, Gram.Type.NGRAM), new Gram("best", 1, Gram.Type.HEAD), new Gram("of", 2, Gram.Type.TAIL) }, {new Gram("of times", 2, Gram.Type.NGRAM), new Gram("of", 2, Gram.Type.HEAD), new Gram("times", 2, Gram.Type.TAIL) }, {new Gram("times the", 1, Gram.Type.NGRAM), new Gram("times", 1, Gram.Type.HEAD), new Gram("the", 1, Gram.Type.TAIL) }, {new Gram("the worst", 1, Gram.Type.NGRAM), new Gram("the", 2, Gram.Type.HEAD), new Gram("worst", 1, Gram.Type.TAIL) }, {new Gram("worst of", 1, Gram.Type.NGRAM), new Gram("worst", 1, Gram.Type.HEAD), new Gram("of", 2, Gram.Type.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} }; Configuration config = getConfiguration(); config.set(LLRReducer.NGRAM_TOTAL, "7"); EasyMock.expect(context.getConfiguration()).andReturn(config); for (int i=0; i < expectations.length; i++) { int[] ee = expectations[i]; context.write(EasyMock.eq(new Text(input[i][0].getString())), (DoubleWritable) EasyMock.anyObject()); EasyMock.expect(ll.logLikelihoodRatio(ee[0], ee[1], ee[2], ee[3])).andDelegateTo(cl); } EasyMock.replay(context, ll); reducer.setup(context); for (Gram[] ii: input) { Collection<Gram> vv = Lists.newLinkedList(); vv.addAll(Arrays.asList(ii).subList(1, ii.length)); reducer.reduce(ii[0], vv, context); } EasyMock.verify(ll); }
@Override protected void reduce(Gram ngram, Iterable<Gram> values, Context context) throws IOException, InterruptedException { int[] gramFreq = {-1, -1}; if (ngram.getType() == Gram.Type.UNIGRAM && emitUnigrams) { DoubleWritable dd = new DoubleWritable(ngram.getFrequency()); Text t = new Text(ngram.getString()); context.write(t, dd); return; } String[] gram = new String[2]; for (Gram value : values) { 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) { context.getCounter(Skipped.EXTRA_HEAD).increment(1); } else { context.getCounter(Skipped.EXTRA_TAIL).increment(1); } return; } gram[pos] = value.getString(); gramFreq[pos] = value.getFrequency(); } if (gramFreq[0] == -1) { log.warn("Missing head for {}, skipping.", ngram); context.getCounter(Skipped.MISSING_HEAD).increment(1); return; } if (gramFreq[1] == -1) { log.warn("Missing tail for {}, skipping", ngram); context.getCounter(Skipped.MISSING_TAIL).increment(1); return; } long k11 = ngram.getFrequency(); long k12 = gramFreq[0] - ngram.getFrequency(); long k21 = gramFreq[1] - ngram.getFrequency(); long k22 = ngramTotal - (gramFreq[0] + gramFreq[1] - ngram.getFrequency()); double llr; try { llr = ll.logLikelihoodRatio(k11, k12, k21, k22); } catch (IllegalArgumentException ex) { context.getCounter(Skipped.LLR_CALCULATION_ERROR).increment(1); log.warn("Problem calculating LLR ratio for ngram {}, HEAD {}:{}, TAIL {}:{}, k11/k12/k21/k22: {}/{}/{}/{}", ngram, gram[0], gramFreq[0], gram[1], gramFreq[1], k11, k12, k21, k22, ex); return; } if (llr < minLLRValue) { context.getCounter(Skipped.LESS_THAN_MIN_LLR).increment(1); } else { context.write(new Text(ngram.getString()), new DoubleWritable(llr)); } }
LLRReducer extends Reducer<Gram, Gram, Text, DoubleWritable> { @Override protected void reduce(Gram ngram, Iterable<Gram> values, Context context) throws IOException, InterruptedException { int[] gramFreq = {-1, -1}; if (ngram.getType() == Gram.Type.UNIGRAM && emitUnigrams) { DoubleWritable dd = new DoubleWritable(ngram.getFrequency()); Text t = new Text(ngram.getString()); context.write(t, dd); return; } String[] gram = new String[2]; for (Gram value : values) { 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) { context.getCounter(Skipped.EXTRA_HEAD).increment(1); } else { context.getCounter(Skipped.EXTRA_TAIL).increment(1); } return; } gram[pos] = value.getString(); gramFreq[pos] = value.getFrequency(); } if (gramFreq[0] == -1) { log.warn("Missing head for {}, skipping.", ngram); context.getCounter(Skipped.MISSING_HEAD).increment(1); return; } if (gramFreq[1] == -1) { log.warn("Missing tail for {}, skipping", ngram); context.getCounter(Skipped.MISSING_TAIL).increment(1); return; } long k11 = ngram.getFrequency(); long k12 = gramFreq[0] - ngram.getFrequency(); long k21 = gramFreq[1] - ngram.getFrequency(); long k22 = ngramTotal - (gramFreq[0] + gramFreq[1] - ngram.getFrequency()); double llr; try { llr = ll.logLikelihoodRatio(k11, k12, k21, k22); } catch (IllegalArgumentException ex) { context.getCounter(Skipped.LLR_CALCULATION_ERROR).increment(1); log.warn("Problem calculating LLR ratio for ngram {}, HEAD {}:{}, TAIL {}:{}, k11/k12/k21/k22: {}/{}/{}/{}", ngram, gram[0], gramFreq[0], gram[1], gramFreq[1], k11, k12, k21, k22, ex); return; } if (llr < minLLRValue) { context.getCounter(Skipped.LESS_THAN_MIN_LLR).increment(1); } else { context.write(new Text(ngram.getString()), new DoubleWritable(llr)); } } }
LLRReducer extends Reducer<Gram, Gram, Text, DoubleWritable> { @Override protected void reduce(Gram ngram, Iterable<Gram> values, Context context) throws IOException, InterruptedException { int[] gramFreq = {-1, -1}; if (ngram.getType() == Gram.Type.UNIGRAM && emitUnigrams) { DoubleWritable dd = new DoubleWritable(ngram.getFrequency()); Text t = new Text(ngram.getString()); context.write(t, dd); return; } String[] gram = new String[2]; for (Gram value : values) { 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) { context.getCounter(Skipped.EXTRA_HEAD).increment(1); } else { context.getCounter(Skipped.EXTRA_TAIL).increment(1); } return; } gram[pos] = value.getString(); gramFreq[pos] = value.getFrequency(); } if (gramFreq[0] == -1) { log.warn("Missing head for {}, skipping.", ngram); context.getCounter(Skipped.MISSING_HEAD).increment(1); return; } if (gramFreq[1] == -1) { log.warn("Missing tail for {}, skipping", ngram); context.getCounter(Skipped.MISSING_TAIL).increment(1); return; } long k11 = ngram.getFrequency(); long k12 = gramFreq[0] - ngram.getFrequency(); long k21 = gramFreq[1] - ngram.getFrequency(); long k22 = ngramTotal - (gramFreq[0] + gramFreq[1] - ngram.getFrequency()); double llr; try { llr = ll.logLikelihoodRatio(k11, k12, k21, k22); } catch (IllegalArgumentException ex) { context.getCounter(Skipped.LLR_CALCULATION_ERROR).increment(1); log.warn("Problem calculating LLR ratio for ngram {}, HEAD {}:{}, TAIL {}:{}, k11/k12/k21/k22: {}/{}/{}/{}", ngram, gram[0], gramFreq[0], gram[1], gramFreq[1], k11, k12, k21, k22, ex); return; } if (llr < minLLRValue) { context.getCounter(Skipped.LESS_THAN_MIN_LLR).increment(1); } else { context.write(new Text(ngram.getString()), new DoubleWritable(llr)); } } LLRReducer();   LLRReducer(LLCallback ll); }
LLRReducer extends Reducer<Gram, Gram, Text, DoubleWritable> { @Override protected void reduce(Gram ngram, Iterable<Gram> values, Context context) throws IOException, InterruptedException { int[] gramFreq = {-1, -1}; if (ngram.getType() == Gram.Type.UNIGRAM && emitUnigrams) { DoubleWritable dd = new DoubleWritable(ngram.getFrequency()); Text t = new Text(ngram.getString()); context.write(t, dd); return; } String[] gram = new String[2]; for (Gram value : values) { 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) { context.getCounter(Skipped.EXTRA_HEAD).increment(1); } else { context.getCounter(Skipped.EXTRA_TAIL).increment(1); } return; } gram[pos] = value.getString(); gramFreq[pos] = value.getFrequency(); } if (gramFreq[0] == -1) { log.warn("Missing head for {}, skipping.", ngram); context.getCounter(Skipped.MISSING_HEAD).increment(1); return; } if (gramFreq[1] == -1) { log.warn("Missing tail for {}, skipping", ngram); context.getCounter(Skipped.MISSING_TAIL).increment(1); return; } long k11 = ngram.getFrequency(); long k12 = gramFreq[0] - ngram.getFrequency(); long k21 = gramFreq[1] - ngram.getFrequency(); long k22 = ngramTotal - (gramFreq[0] + gramFreq[1] - ngram.getFrequency()); double llr; try { llr = ll.logLikelihoodRatio(k11, k12, k21, k22); } catch (IllegalArgumentException ex) { context.getCounter(Skipped.LLR_CALCULATION_ERROR).increment(1); log.warn("Problem calculating LLR ratio for ngram {}, HEAD {}:{}, TAIL {}:{}, k11/k12/k21/k22: {}/{}/{}/{}", ngram, gram[0], gramFreq[0], gram[1], gramFreq[1], k11, k12, k21, k22, ex); return; } if (llr < minLLRValue) { context.getCounter(Skipped.LESS_THAN_MIN_LLR).increment(1); } else { context.write(new Text(ngram.getString()), new DoubleWritable(llr)); } } LLRReducer();   LLRReducer(LLCallback ll);  }
LLRReducer extends Reducer<Gram, Gram, Text, DoubleWritable> { @Override protected void reduce(Gram ngram, Iterable<Gram> values, Context context) throws IOException, InterruptedException { int[] gramFreq = {-1, -1}; if (ngram.getType() == Gram.Type.UNIGRAM && emitUnigrams) { DoubleWritable dd = new DoubleWritable(ngram.getFrequency()); Text t = new Text(ngram.getString()); context.write(t, dd); return; } String[] gram = new String[2]; for (Gram value : values) { 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) { context.getCounter(Skipped.EXTRA_HEAD).increment(1); } else { context.getCounter(Skipped.EXTRA_TAIL).increment(1); } return; } gram[pos] = value.getString(); gramFreq[pos] = value.getFrequency(); } if (gramFreq[0] == -1) { log.warn("Missing head for {}, skipping.", ngram); context.getCounter(Skipped.MISSING_HEAD).increment(1); return; } if (gramFreq[1] == -1) { log.warn("Missing tail for {}, skipping", ngram); context.getCounter(Skipped.MISSING_TAIL).increment(1); return; } long k11 = ngram.getFrequency(); long k12 = gramFreq[0] - ngram.getFrequency(); long k21 = gramFreq[1] - ngram.getFrequency(); long k22 = ngramTotal - (gramFreq[0] + gramFreq[1] - ngram.getFrequency()); double llr; try { llr = ll.logLikelihoodRatio(k11, k12, k21, k22); } catch (IllegalArgumentException ex) { context.getCounter(Skipped.LLR_CALCULATION_ERROR).increment(1); log.warn("Problem calculating LLR ratio for ngram {}, HEAD {}:{}, TAIL {}:{}, k11/k12/k21/k22: {}/{}/{}/{}", ngram, gram[0], gramFreq[0], gram[1], gramFreq[1], k11, k12, k21, k22, ex); return; } if (llr < minLLRValue) { context.getCounter(Skipped.LESS_THAN_MIN_LLR).increment(1); } else { context.write(new Text(ngram.getString()), new DoubleWritable(llr)); } } LLRReducer();   LLRReducer(LLCallback ll);  static final String NGRAM_TOTAL; static final String MIN_LLR; static final float DEFAULT_MIN_LLR; }