1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.util;
21
22 import java.util.HashSet;
23 import java.util.Map;
24 import java.util.Set;
25 import java.util.concurrent.ConcurrentHashMap;
26
27 import org.apache.hadoop.conf.Configuration;
28 import org.apache.hadoop.hbase.HBaseConfiguration;
29 import org.apache.hadoop.hbase.HConstants;
30 import org.apache.hadoop.hbase.classification.InterfaceAudience;
31 import org.slf4j.Logger;
32 import org.slf4j.LoggerFactory;
33
34
35
36
37
38
39
40
41
42
43
44
45
46 @InterfaceAudience.Public
47 public class LossyCounting {
48 private static final Logger LOG = LoggerFactory.getLogger(LossyCounting.class);
49 private long bucketSize;
50 private long currentTerm;
51 private double errorRate;
52 private Map<String, Integer> data;
53 private long totalDataCount;
54
55 public LossyCounting(double errorRate) {
56 this.errorRate = errorRate;
57 if (errorRate < 0.0 || errorRate > 1.0) {
58 throw new IllegalArgumentException(" Lossy Counting error rate should be within range [0,1]");
59 }
60 this.bucketSize = (long) Math.ceil(1 / errorRate);
61 this.currentTerm = 1;
62 this.totalDataCount = 0;
63 this.errorRate = errorRate;
64 this.data = new ConcurrentHashMap<>();
65 calculateCurrentTerm();
66 }
67
68 public LossyCounting() {
69 Configuration conf = HBaseConfiguration.create();
70 this.errorRate = conf.getDouble(HConstants.DEFAULT_LOSSY_COUNTING_ERROR_RATE, 0.02);
71 this.bucketSize = (long) Math.ceil(1.0 / errorRate);
72 this.currentTerm = 1;
73 this.totalDataCount = 0;
74 this.data = new ConcurrentHashMap<>();
75 calculateCurrentTerm();
76 }
77
78 public Set<String> addByOne(String key) {
79 if(data.containsKey(key)) {
80 data.put(key, data.get(key) +1);
81 } else {
82 data.put(key, 1);
83 }
84 totalDataCount++;
85 calculateCurrentTerm();
86 Set<String> dataToBeSwept = new HashSet<>();
87 if(totalDataCount % bucketSize == 0) {
88 dataToBeSwept = sweep();
89 }
90 return dataToBeSwept;
91 }
92
93
94
95
96
97 private Set<String> sweep() {
98 Set<String> dataToBeSwept = new HashSet<>();
99 for(Map.Entry<String, Integer> entry : data.entrySet()) {
100 if(entry.getValue() + errorRate < currentTerm) {
101 dataToBeSwept.add(entry.getKey());
102 }
103 }
104 for(String key : dataToBeSwept) {
105 data.remove(key);
106 }
107 LOG.debug(String.format("Swept %d of elements.", dataToBeSwept.size()));
108 return dataToBeSwept;
109 }
110
111
112
113
114 private void calculateCurrentTerm() {
115 this.currentTerm = (int) Math.ceil(1.0 * totalDataCount / bucketSize);
116 }
117
118 public long getBuketSize(){
119 return bucketSize;
120 }
121
122 public long getDataSize() {
123 return data.size();
124 }
125
126 public boolean contains(String key) {
127 return data.containsKey(key);
128 }
129
130 public long getCurrentTerm() {
131 return currentTerm;
132 }
133 }
134