CountMinHeavyHitter implements HeavyHitter, Serializable { public void merge(HeavyHitter toMerge) throws CMHeavyHitterMergeException { try { CountMinHeavyHitter cmToMerge = (CountMinHeavyHitter)toMerge; if (this.fraction != cmToMerge.fraction) { throw new CMHeavyHitterMergeException("The fraction for both heavy hitters must be the same"); } this.countMinSketch = CountMinSketch.merge(this.countMinSketch, cmToMerge.countMinSketch); HashMap<Object,Long> mergedHeavyHitters = new HashMap<Object, Long>(); for (Map.Entry<Object, Long> entry : this.heavyHitters.entrySet()) { mergedHeavyHitters.put(entry.getKey(), estimateCount(entry.getKey())); } for (Map.Entry<Object, Long> entry : cmToMerge.heavyHitters.entrySet()) { if (!mergedHeavyHitters.containsKey(entry.getKey())) { mergedHeavyHitters.put(entry.getKey(), estimateCount(entry.getKey())); } } this.heavyHitters = mergedHeavyHitters; cardinality+=cmToMerge.cardinality; }catch (ClassCastException ex){ throw new CMHeavyHitterMergeException("Both heavy hitter objects must belong to the same class"); }catch (Exception ex){ throw new CMHeavyHitterMergeException("Cannot merge count min sketches: "+ex.getMessage()); } } CountMinHeavyHitter(double fraction, double error, double confidence, int seed);  CountMinHeavyHitter(CountMinSketch countMinSketch, double fraction); @Override void addObject(Object o); long estimateCount(Object item); void merge(HeavyHitter toMerge); @Override HashMap<Object,Long> getHeavyHitters(); @Override String toString();  }
@Test public void merge() throws CountMinHeavyHitter.CMHeavyHitterMergeException { int numToMerge = 5; long[] actualFreq = new long[maxScale]; CountMinHeavyHitter merged = new CountMinHeavyHitter(fraction,error,confidence,seed); long totalCardinality = 0; CountMinHeavyHitter[] sketches = new CountMinHeavyHitter[numToMerge]; for (int i = 0; i < numToMerge; i++) { CountMinSketch cms = new CountMinSketch(error, confidence, seed); sketches[i] = new CountMinHeavyHitter(cms, fraction); for (int j = 0; j < cardinality; j++) { int val; if (r.nextDouble()<0.1){ val = r.nextInt(10); }else{ val = r.nextInt(maxScale); } sketches[i].addObject(val); actualFreq[val]++; totalCardinality++; } merged.merge(sketches[i]); } Map<Object,Long> mergedHeavyHitters = merged.getHeavyHitters(); long frequency = (long)(totalCardinality*fraction); for (int i = 0; i < actualFreq.length; ++i) { if (actualFreq[i] >= frequency) { assertTrue("All items with freq. > s.n will be found. Item " + i + ". Real freq. " + actualFreq[i] + " Expected freq." + frequency, mergedHeavyHitters.containsKey(i)); } } }