LossyCounting implements HeavyHitter, Serializable { public void merge(HeavyHitter toMerge) throws HeavyHitterMergeException { try{ LossyCounting lsToMerge = (LossyCounting)toMerge; if (this.fraction!=lsToMerge.fraction){ throw new HeavyHitterMergeException("Both heavy hitter structures must be identical"); } this.cardinality+=lsToMerge.cardinality; this.bucket = (long)Math.floor(cardinality*error); for (Map.Entry<Object, Counter> entry : lsToMerge.heavyHitters.entrySet()){ Counter counter = this.heavyHitters.get(entry.getKey()); if (counter==null){ this.heavyHitters.put(entry.getKey(),entry.getValue()); }else{ Counter mergingCounter = entry.getValue(); this.heavyHitters.put(entry.getKey(), new Counter(mergingCounter.lowerBound+counter.lowerBound, mergingCounter.frequencyError +counter.frequencyError)); } } updateHeavyHitters(); }catch (ClassCastException ex){ throw new HeavyHitterMergeException("Both heavy hitter structures must be identical"); } } LossyCounting(double fraction, double error); @Override void addObject(Object o); void updateHeavyHitters(); void merge(HeavyHitter toMerge); @Override HashMap<Object,Long> getHeavyHitters(); @Override String toString();  }
@Test public void merge() throws HeavyHitterMergeException { int numToMerge = 5; LossyCounting merged = new LossyCounting(fraction,error); LossyCounting[] sketches = new LossyCounting[numToMerge]; long[] actualFreq = new long[maxScale]; long totalCardinality = 0; for (int i = 0; i < numToMerge; i++) { sketches[i] = new LossyCounting(fraction,error); 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]); } System.out.println("\nMERGED\n" + merged.toString()); Map<Object,Long> mergedHeavyHitters = merged.getHeavyHitters(); long frequency = (long)(totalCardinality*fraction); long minFrequency = (long)(totalCardinality*(fraction-error)); System.out.println("Frequency Threshold:" + frequency); System.out.println("False positive Threshold:" + minFrequency); System.out.println("Frequency of 14:" + actualFreq[14]); System.out.println("Real frequent items: "); for (int i = 0; i < actualFreq.length; ++i) { if (actualFreq[i]>=frequency) { System.out.println(i+": "+actualFreq[i]); assertTrue("All items with freq. > s.n will be found. Item " + i + ". Real freq. " + actualFreq[i]+" Expected freq."+ frequency, mergedHeavyHitters.containsKey(i)); } if (mergedHeavyHitters.containsKey(i)){ assertTrue("no item with freq. < (s-e).n will be found. Item " + i + ". Real freq. " + actualFreq[i]+" Min freq."+ minFrequency, actualFreq[i]>=minFrequency); assertTrue("the estimated freq. underestimates the true freq. by < e.n. Real freq. " + actualFreq[i] + " Lower bound "+mergedHeavyHitters.get(i), Math.abs(mergedHeavyHitters.get(i)-actualFreq[i]) < error*cardinality); } } }