View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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   * LossyCounting utility, bounded data structure that maintains approximate high frequency
37   * elements in data stream.
38   *
39   * Bucket size is 1 / error rate.  (Error rate is 0.02 by default)
40   * Lemma If element does not appear in set, then is frequency is less than e * N
41   *       (N is total element counts until now.)
42   * Based on paper:
43   * http://www.vldb.org/conf/2002/S10P03.pdf
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     * sweep low frequency data
95     * @return Names of elements got swept
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    * Calculate and set current term
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