@Test public void testSummarizing() { SummarizerConfiguration sc = SummarizerConfiguration.builder(FamilySummarizer.class) .addOptions(MAX_COUNTERS_OPT, "5", MAX_COUNTER_LEN_OPT, "10").build(); FamilySummarizer countSum = new FamilySummarizer(); Value val = new Value("abc"); Summarizer.Collector collector = countSum.collector(sc); for (String fam : Arrays.asList("f1", "f1", "f1", "f2", "f1", "f70000000000000000000", "f70000000000000000001", "f2", "f3", "f4", "f5", "f6", "f7", "f3", "f7")) { collector.accept(new Key("r", fam), val); } Key dk = new Key("r", "f2"); dk.setDeleted(true); collector.accept(dk, new Value("")); HashMap<String,Long> stats = new HashMap<>(); collector.summarize(stats::put); String p = COUNTER_STAT_PREFIX; HashMap<String,Long> expected = new HashMap<>(); expected.put(p + "f1", 4L); expected.put(p + "f2", 2L); expected.put(p + "f3", 2L); expected.put(p + "f4", 1L); expected.put(p + "f5", 1L); expected.put(TOO_LONG_STAT, 2L); expected.put(TOO_MANY_STAT, 3L); expected.put(SEEN_STAT, 16L); expected.put(EMITTED_STAT, 15L); expected.put(DELETES_IGNORED_STAT, 1L); assertEquals(expected, stats); CounterSummary csum = new CounterSummary(stats); assertEquals(5, csum.getIgnored()); assertEquals(3, csum.getTooMany()); assertEquals(2, csum.getTooLong()); assertEquals(16, csum.getSeen()); assertEquals(15, csum.getEmitted()); assertEquals(1, csum.getDeletesIgnored()); expected.clear(); expected.put("f1", 4L); expected.put("f2", 2L); expected.put("f3", 2L); expected.put("f4", 1L); expected.put("f5", 1L); assertEquals(expected, csum.getCounters()); }
@Override public Collector collector(SummarizerConfiguration sc) { init(sc); return new Collector() { private Map<K,MutableLong> counters = new HashMap<>(); private long tooMany = 0; private long tooLong = 0; private long seen = 0; private long emitted = 0; private long deleted = 0; private Converter<K> converter = converter(); private Function<K,String> encoder = encoder(); private UnaryOperator<K> copier = copier(); private void incrementCounter(K counter) { emitted++; MutableLong ml = counters.get(counter); if (ml == null) { if (counters.size() >= maxCounters) { tooMany++; } else { if (encoder.apply(counter).length() >= maxCounterKeyLen) { tooLong++; } else { counters.put(copier.apply(counter), new MutableLong(1)); } } } else { ml.increment(); } } @Override public void accept(Key k, Value v) { seen++; if (ignoreDeletes && k.isDeleted()) { deleted++; } else { converter.convert(k, v, this::incrementCounter); } } @Override public void summarize(StatisticConsumer sc) { StringBuilder sb = new StringBuilder(COUNTER_STAT_PREFIX); for (Entry<K,MutableLong> entry : counters.entrySet()) { sb.setLength(COUNTER_STAT_PREFIX.length()); sb.append(encoder.apply(entry.getKey())); sc.accept(sb.toString(), entry.getValue().longValue()); } sc.accept(TOO_MANY_STAT, tooMany); sc.accept(TOO_LONG_STAT, tooLong); sc.accept(EMITTED_STAT, emitted); sc.accept(SEEN_STAT, seen); sc.accept(DELETES_IGNORED_STAT, deleted); } }; }
CountingSummarizer implements Summarizer { @Override public Collector collector(SummarizerConfiguration sc) { init(sc); return new Collector() { private Map<K,MutableLong> counters = new HashMap<>(); private long tooMany = 0; private long tooLong = 0; private long seen = 0; private long emitted = 0; private long deleted = 0; private Converter<K> converter = converter(); private Function<K,String> encoder = encoder(); private UnaryOperator<K> copier = copier(); private void incrementCounter(K counter) { emitted++; MutableLong ml = counters.get(counter); if (ml == null) { if (counters.size() >= maxCounters) { tooMany++; } else { if (encoder.apply(counter).length() >= maxCounterKeyLen) { tooLong++; } else { counters.put(copier.apply(counter), new MutableLong(1)); } } } else { ml.increment(); } } @Override public void accept(Key k, Value v) { seen++; if (ignoreDeletes && k.isDeleted()) { deleted++; } else { converter.convert(k, v, this::incrementCounter); } } @Override public void summarize(StatisticConsumer sc) { StringBuilder sb = new StringBuilder(COUNTER_STAT_PREFIX); for (Entry<K,MutableLong> entry : counters.entrySet()) { sb.setLength(COUNTER_STAT_PREFIX.length()); sb.append(encoder.apply(entry.getKey())); sc.accept(sb.toString(), entry.getValue().longValue()); } sc.accept(TOO_MANY_STAT, tooMany); sc.accept(TOO_LONG_STAT, tooLong); sc.accept(EMITTED_STAT, emitted); sc.accept(SEEN_STAT, seen); sc.accept(DELETES_IGNORED_STAT, deleted); } }; } }
CountingSummarizer implements Summarizer { @Override public Collector collector(SummarizerConfiguration sc) { init(sc); return new Collector() { private Map<K,MutableLong> counters = new HashMap<>(); private long tooMany = 0; private long tooLong = 0; private long seen = 0; private long emitted = 0; private long deleted = 0; private Converter<K> converter = converter(); private Function<K,String> encoder = encoder(); private UnaryOperator<K> copier = copier(); private void incrementCounter(K counter) { emitted++; MutableLong ml = counters.get(counter); if (ml == null) { if (counters.size() >= maxCounters) { tooMany++; } else { if (encoder.apply(counter).length() >= maxCounterKeyLen) { tooLong++; } else { counters.put(copier.apply(counter), new MutableLong(1)); } } } else { ml.increment(); } } @Override public void accept(Key k, Value v) { seen++; if (ignoreDeletes && k.isDeleted()) { deleted++; } else { converter.convert(k, v, this::incrementCounter); } } @Override public void summarize(StatisticConsumer sc) { StringBuilder sb = new StringBuilder(COUNTER_STAT_PREFIX); for (Entry<K,MutableLong> entry : counters.entrySet()) { sb.setLength(COUNTER_STAT_PREFIX.length()); sb.append(encoder.apply(entry.getKey())); sc.accept(sb.toString(), entry.getValue().longValue()); } sc.accept(TOO_MANY_STAT, tooMany); sc.accept(TOO_LONG_STAT, tooLong); sc.accept(EMITTED_STAT, emitted); sc.accept(SEEN_STAT, seen); sc.accept(DELETES_IGNORED_STAT, deleted); } }; }  }
CountingSummarizer implements Summarizer { @Override public Collector collector(SummarizerConfiguration sc) { init(sc); return new Collector() { private Map<K,MutableLong> counters = new HashMap<>(); private long tooMany = 0; private long tooLong = 0; private long seen = 0; private long emitted = 0; private long deleted = 0; private Converter<K> converter = converter(); private Function<K,String> encoder = encoder(); private UnaryOperator<K> copier = copier(); private void incrementCounter(K counter) { emitted++; MutableLong ml = counters.get(counter); if (ml == null) { if (counters.size() >= maxCounters) { tooMany++; } else { if (encoder.apply(counter).length() >= maxCounterKeyLen) { tooLong++; } else { counters.put(copier.apply(counter), new MutableLong(1)); } } } else { ml.increment(); } } @Override public void accept(Key k, Value v) { seen++; if (ignoreDeletes && k.isDeleted()) { deleted++; } else { converter.convert(k, v, this::incrementCounter); } } @Override public void summarize(StatisticConsumer sc) { StringBuilder sb = new StringBuilder(COUNTER_STAT_PREFIX); for (Entry<K,MutableLong> entry : counters.entrySet()) { sb.setLength(COUNTER_STAT_PREFIX.length()); sb.append(encoder.apply(entry.getKey())); sc.accept(sb.toString(), entry.getValue().longValue()); } sc.accept(TOO_MANY_STAT, tooMany); sc.accept(TOO_LONG_STAT, tooLong); sc.accept(EMITTED_STAT, emitted); sc.accept(SEEN_STAT, seen); sc.accept(DELETES_IGNORED_STAT, deleted); } }; }  @Override Collector collector(SummarizerConfiguration sc); @Override Combiner combiner(SummarizerConfiguration sc); }
CountingSummarizer implements Summarizer { @Override public Collector collector(SummarizerConfiguration sc) { init(sc); return new Collector() { private Map<K,MutableLong> counters = new HashMap<>(); private long tooMany = 0; private long tooLong = 0; private long seen = 0; private long emitted = 0; private long deleted = 0; private Converter<K> converter = converter(); private Function<K,String> encoder = encoder(); private UnaryOperator<K> copier = copier(); private void incrementCounter(K counter) { emitted++; MutableLong ml = counters.get(counter); if (ml == null) { if (counters.size() >= maxCounters) { tooMany++; } else { if (encoder.apply(counter).length() >= maxCounterKeyLen) { tooLong++; } else { counters.put(copier.apply(counter), new MutableLong(1)); } } } else { ml.increment(); } } @Override public void accept(Key k, Value v) { seen++; if (ignoreDeletes && k.isDeleted()) { deleted++; } else { converter.convert(k, v, this::incrementCounter); } } @Override public void summarize(StatisticConsumer sc) { StringBuilder sb = new StringBuilder(COUNTER_STAT_PREFIX); for (Entry<K,MutableLong> entry : counters.entrySet()) { sb.setLength(COUNTER_STAT_PREFIX.length()); sb.append(encoder.apply(entry.getKey())); sc.accept(sb.toString(), entry.getValue().longValue()); } sc.accept(TOO_MANY_STAT, tooMany); sc.accept(TOO_LONG_STAT, tooLong); sc.accept(EMITTED_STAT, emitted); sc.accept(SEEN_STAT, seen); sc.accept(DELETES_IGNORED_STAT, deleted); } }; }  @Override Collector collector(SummarizerConfiguration sc); @Override Combiner combiner(SummarizerConfiguration sc); static final String MAX_COUNTERS_OPT; static final String MAX_COUNTER_LEN_OPT; static final String INGNORE_DELETES_OPT; static final String COUNTER_STAT_PREFIX; static final String TOO_MANY_STAT; static final String TOO_LONG_STAT; static final String EMITTED_STAT; static final String DELETES_IGNORED_STAT; static final String SEEN_STAT; static final String MAX_COUNTER_DEFAULT; static final String MAX_CKL_DEFAULT; static final String INGNORE_DELETES_DEFAULT; }