「随机算法专题」Count-min Sketch 入门

Point Query 问题：给出$n$个数$a_1, \ldots, a_n$，$0\le a_i\le M$，你需要（近似）回答$x$的出现次数。

容易想到直接开一个桶，空间复杂度$O(n)$。我们希望在$n$很大时用更小的空间复杂度解决这个问题。

Count-min Sketch

Count-min Sketch 是一个数据结构，通过随机哈希的方式来给出上述问题的近似解。

考虑构造一个随机哈希函数$h: [M] \to [m]$，将输入的元素均匀映射到$m$个 bucket。

利用这个哈希来计算每个哈希值的出现次数：$C[x] = \sum_i P[h(a_i) = x]$。那么记$x$的真实出现次数为$c_x = \sum_i[a_i = x]$，显然有$C[h(x)] \ge c_x$，在哈希不冲突时取等。

容易想到，多次采用不同的随机哈希$h_1, \ldots, h_t$分别统计出现次数$C_1, \ldots, C_t$，那么取$\min_{1\le j\le t} C_j[h_j(x)]$可以很好地逼近$c_x$。

对此进行更细致的期望分析：

$$\begin{align} E\big( C[h(x)]\big) &= E\left(\sum_i [h(a_i) = h(x)] \right) \\ &= E\left(c_x + \sum_{y\in [M], y\ne x} [h(y) = h(x)] c_y \right) \\ &= c_x + \sum_{y\in [M], y\ne x} E\big( [h(y) = h(x)] \big) c_y \\ &= c_x + \sum_{y\in [M], y\ne x} \text{Pr}[h(y) = h(x)]c_y \\ \end{align}$$

假设哈希函数是均匀随机，那么$\text{Pr}[h(y) = h(x)] = \frac{1}{m}$，于是

$$E\big( C[h(x)]\big) = c_x + \frac{1}{m} (n - c_x) \le c_x + \frac{n}{m}$$

于是我们对该随机算法进行误差估计：

$$\text{Pr}\big[ C[h(x)] - c_x > n\varepsilon \big] \le \frac{E\big( C[h(x)] - c_x \big)}{n\varepsilon} = \frac{1}{m\varepsilon}$$

这里面用到了 Markov Inequality：若$X\ge 0$，有$a\cdot \text{Pr}[X\ge a] \le E(X)$。

取$m = \frac{2}{\varepsilon}$可以控制上述概率不大于$\frac{1}{2}$，那么我们取$O(\log_2n)$个独立随机哈希（若已知$n$的上界），即可求出绝对误差在$n\varepsilon$内的近似解。

空间复杂度$O(mt) = O(m\log_2n)$。

K-Heavy Hitter

K-Heavy Hitter 问题：给出$n$个数$a_1, \ldots, a_n$，$0\le a_i\le M$，你需要（近似）求出出现次数前$k$多的所有数。

对于上述问题我们仍然有空间复杂度$O(n)$的朴素算法。

近似 K-Heavy Hitter：要求必须求出所有出现次数不少于$\frac{n}{k}$的数，并且求出的所有数的出现次数不少于$\frac{n}{k} - n\varepsilon$。

若已知$n$，那么求近似解可以直接使用 Count-min Sketch 解决。考虑$n$上界未知但$\log_2n$上界已知的情况。

我们维护当前插入的总元素个数$n'$。在 Count-min Sketch 的基础上我们额外维护一个集合$L$，用于记录 Count-min Sketch 意义下出现次数不小于$\frac{n'}{k}$的数。每次插入元素后，我们将出现次数小于$\frac{n'}{k}$的数从中删除。

可以发现近似 K-Heavy Hitter 的误差主要是留给 Count-min Sketch。而上述$L$集合的维护其实是足够准确的。

由于出现次数不小于$\frac{n}{k}$的数字个数是$O(k)$的，因此$L$的空间大小有保证。在考虑到 Count-min Sketch 的误差后可以得出该算法的空间复杂度为$O(m\log_2n + k + \frac{1}{\varepsilon})$。

一个可能的代码实现：

class Solver {
public:
#define FOR(a, b, c) for(int a = (int)(b); a <= (int)(c); a++)
  int n1 = 0, k = 0; // current total numbers
  int* L, len = 0;

  static const int T = 6; // number of hash tables
  // const double eps = 0.01;
  static const int M = (5 * (1 << 20)) / T; // 2 / eps + 11; // size of hash table

  struct hash_table {
    int seed;
    int* bucket;

    hash_table() {}
    hash_table(int seed, int* bucket): seed(seed), bucket(bucket) {
      FOR(i, 0, M - 1) bucket[i] = 0;
    }
    void add(int val) {
      bucket[fasthash64(val, seed) % M]++;
    }
    int get(int val) {
      return bucket[fasthash64(val, seed) % M];
    }
  };
  hash_table h[T];

  int count_min(int e) {
    int res = h[0].get(e);
    FOR(i, 1, T - 1) res = min(res, h[i].get(e));
    return res;
  }
  // arr 是题目提供的内存空间，约为 32MB
  void init(int* arr, int m, int _k) {
    k = _k;
    FOR(i, 0, T - 1) {
      h[i] = hash_table(1145140 + i, arr + M * i);
    }
    L = arr + M * T;
  }
  void add(int e) {
    ++n1;
    FOR(i, 0, T - 1) h[i].add(e);

    FOR(i, 0, len - 1) {
      if (count_min(L[i]) < n1 * 1.0 / k) { // delete it
        L[i] = L[len - 1];
        --len, --i;
      }
    }

    if (count_min(e) >= n1 * 1.0 / k) {
      FOR(i, 0, len - 1) {
        if (L[i] == e) return;
      }
      L[len++] = e;
    }
  }
  std::vector<int> report() {
    return vector<int>(L, L + len);
  }
};