算法引入
在大数据实时处理场景中，一个常见的需求是：如何在有限内存下快速统计海量数据流中各元素的出现频率？ 传统哈希表需要O(n)空间，无法应对数十亿级别的数据量。Count-Min Sketch（CMS）由Graham Cormode和S. Muthukrishnan于2005年提出，是一种概率性数据结构，以极小的空间开销和O(1)查询时间提供元素频率的近似估计，广泛应用于网络流量监控、热门商品统计、数据库查询优化等场景。
核心思想
Count-Min Sketch的核心思想与布隆过滤器类似：使用多个独立的哈希函数将元素映射到一个二维计数矩阵的不同位置。当增加某个元素的计数时，对所有对应的矩阵位置都加1；查询时取所有映射位置的最小值作为估计频率。由于哈希冲突只会导致计数被高估（overestimate），而取最小值能有效降低这种高估的偏差。
结构定义
- 一个宽度为
w、深度为d的二维计数数组count[d][w] d个两两独立的哈希函数h₁, h₂, ..., h_d- 宽度
w = ⌈e/ε⌉，深度d = ⌈ln(1/δ)⌉，其中 ε 为误差率，δ 为失败概率
操作流程
- 增加计数（Add）：对每个哈希函数
h_i，计算count[i][h_i(x) % w] += 1 - 查询频率（Query）：返回
min(count[i][h_i(x) % w])，其中 i = 1..d - 误差保证：估计值
f̂(x)满足f(x) ≤ f̂(x) ≤ f(x) + ε·N的概率至少为1-δ
Java完整实现
import java.util.Random;
/**
* Count-Min Sketch概率性频率统计数据结构
* 适用于海量数据流中的元素频率近似查询
*/
public class CountMinSketch {
private final int width; // 计数矩阵的宽度
private final int depth; // 哈希函数的数量（深度）
private final long[][] count; // 二维计数矩阵
private final long[] hashA; // 哈希函数线性参数a
private final long[] hashB; // 哈希函数线性参数b
private final long prime; // 大素数，用于哈希计算
private long totalCount; // 总元素计数
/**
* 构造函数：根据误差率ε和置信度δ自动计算矩阵尺寸
* @param epsilon 误差率，如0.001表示估计误差不超过0.1%的总计数
* @param delta 失败概率，如0.01表示99%的查询满足误差界
*/
public CountMinSketch(double epsilon, double delta) {
this.width = (int) Math.ceil(Math.E / epsilon);
this.depth = (int) Math.ceil(Math.log(1.0 / delta));
this.count = new long[depth][width];
this.hashA = new long[depth];
this.hashB = new long[depth];
this.prime = 2147483647L; // 大素数 2^31 - 1
this.totalCount = 0;
// 初始化两两独立的哈希函数参数
Random rand = new Random(42); // 固定种子保证可重复性
for (int i = 0; i < depth; i++) {
hashA[i] = rand.nextInt((int) prime - 1) + 1;
hashB[i] = rand.nextInt((int) prime);
}
}
/**
* 构造函数：直接指定矩阵尺寸
*/
public CountMinSketch(int width, int depth) {
this.width = width;
this.depth = depth;
this.count = new long[depth][width];
this.hashA = new long[depth];
this.hashB = new long[depth];
this.prime = 2147483647L;
this.totalCount = 0;
Random rand = new Random(42);
for (int i = 0; i < depth; i++) {
hashA[i] = rand.nextInt((int) prime - 1) + 1;
hashB[i] = rand.nextInt((int) prime);
}
}
/**
* 将字符串元素映射为哈希值
* 使用FNV-1a哈希作为基础
*/
private long hashString(String item) {
long hash = 0x811c9dc5L;
for (char c : item.toCharArray()) {
hash ^= c;
hash *= 0x01000193L;
}
return Math.abs(hash);
}
/**
* 第i个哈希函数：h_i(x) = (a_i * x + b_i) % prime % width
*/
private int hash(int i, long itemHash) {
long result = (hashA[i] * itemHash + hashB[i]) % prime;
return (int) (result % width);
}
/**
* 增加元素的计数
* @param item 要计数的元素（字符串）
* @param delta 增加量，通常为1
*/
public void add(String item, long delta) {
long itemHash = hashString(item);
for (int i = 0; i < depth; i++) {
int index = hash(i, itemHash);
count[i][index] += delta;
}
totalCount += delta;
}
/**
* 增加元素计数（默认增量为1）
*/
public void add(String item) {
add(item, 1);
}
/**
* 查询元素的估计频率
* 返回所有哈希位置的最小值，保证不会低估
* @param item 要查询的元素
* @return 估计频率（可能略高于真实值）
*/
public long estimate(String item) {
long itemHash = hashString(item);
long minCount = Long.MAX_VALUE;
for (int i = 0; i < depth; i++) {
int index = hash(i, itemHash);
minCount = Math.min(minCount, count[i][index]);
}
return minCount;
}
/**
* 查询元素频率占比（估计值 / 总计数）
*/
public double estimateFrequency(String item) {
if (totalCount == 0) return 0.0;
return (double) estimate(item) / totalCount;
}
/**
* 获取数据流中的总元素数量
*/
public long getTotalCount() {
return totalCount;
}
/**
* 获取Top-K频繁元素（使用辅助最小堆）
* 注意：这是近似结果，基于CMS的估计值
*/
public java.util.List<String> topK(java.util.Set<String> candidates, int k) {
// 对所有候选元素计算估计频率
java.util.PriorityQueue<String[]> minHeap =
new java.util.PriorityQueue<>(k, (a, b) -> Long.compare(
Long.parseLong(a[1]), Long.parseLong(b[1])
));
for (String item : candidates) {
long freq = estimate(item);
if (minHeap.size() < k) {
minHeap.offer(new String[]{item, String.valueOf(freq)});
} else if (freq > Long.parseLong(minHeap.peek()[1])) {
minHeap.poll();
minHeap.offer(new String[]{item, String.valueOf(freq)});
}
}
java.util.List<String> result = new java.util.ArrayList<>();
while (!minHeap.isEmpty()) {
result.add(minHeap.poll()[0]);
}
java.util.Collections.reverse(result);
return result;
}
/**
* 合并两个CMS（要求尺寸和哈希参数相同）
*/
public void merge(CountMinSketch other) {
if (this.width != other.width || this.depth != other.depth) {
throw new IllegalArgumentException("CMS尺寸不匹配，无法合并");
}
for (int i = 0; i < depth; i++) {
for (int j = 0; j < width; j++) {
this.count[i][j] += other.count[i][j];
}
}
this.totalCount += other.totalCount;
}
/**
* 打印矩阵统计信息
*/
public void printStats() {
System.out.println("Count-Min Sketch 统计:");
System.out.println(" 矩阵尺寸: " + depth + " x " + width);
System.out.println(" 总元素数: " + totalCount);
System.out.println(" 空间占用: " + (depth * width * 8 / 1024.0) + " KB");
long nonZero = 0;
long maxCount = 0;
for (int i = 0; i < depth; i++) {
for (int j = 0; j < width; j++) {
if (count[i][j] > 0) nonZero++;
maxCount = Math.max(maxCount, count[i][j]);
}
}
System.out.println(" 非零单元: " + nonZero + " / " + (depth * width));
System.out.println(" 最大计数: " + maxCount);
}
public static void main(String[] args) {
// 创建CMS：误差率0.1%，置信度99.9%
CountMinSketch cms = new CountMinSketch(0.001, 0.001);
// 模拟热门商品点击流
String[] products = {
"iPhone16", "Mate70", "iPhone16", "iPadPro",
"Mate70", "iPhone16", "MacBook", "Mate70",
"iPhone16", "AirPods", "Mate70", "iPhone16",
"MacBook", "Mate70", "iPhone16", "Watch",
"iPhone16", "Mate70", "iPhone16", "Mate70"
};
// 模拟大量数据流
Random rand = new Random(123);
for (int i = 0; i < 100000; i++) {
// iPhone16 出现概率40%
if (rand.nextDouble() < 0.40) {
cms.add("iPhone16");
} else if (rand.nextDouble() < 0.30) {
cms.add("Mate70");
} else if (rand.nextDouble() < 0.15) {
cms.add("MacBook");
} else {
String[] others = {"AirPods", "Watch", "iPadPro", "Pencil", "MagSafe"};
cms.add(others[rand.nextInt(others.length)]);
}
}
cms.printStats();
System.out.println("\n频率估计结果:");
String[] queries = {"iPhone16", "Mate70", "MacBook", "AirPods", "Watch", "HuaweiPad"};
for (String q : queries) {
long est = cms.estimate(q);
double ratio = cms.estimateFrequency(q) * 100;
System.out.printf(" %s: 估计计数=%d, 占比=%.2f%%%n", q, est, ratio);
}
// Top-K查询
java.util.Set<String> candidates = java.util.Set.of(
"iPhone16", "Mate70", "MacBook", "AirPods", "Watch", "iPadPro", "Pencil", "MagSafe"
);
System.out.println("\nTop-3 热门商品: " + cms.topK(candidates, 3));
}
}
算法优化与变体
Conservative Update
标准的CMS在Add操作时对所有哈希位置都加1。Conservative Update优化只增加当前估计值最小的那些位置，减少哈希冲突导致的计数膨胀：
public void addConservative(String item, long delta) {
long itemHash = hashString(item);
long minCount = Long.MAX_VALUE;
int[] indices = new int[depth];
for (int i = 0; i < depth; i++) {
indices[i] = hash(i, itemHash);
minCount = Math.min(minCount, count[i][indices[i]]);
}
for (int i = 0; i < depth; i++) {
if (count[i][indices[i]] == minCount) {
count[i][indices[i]] += delta;
}
}
totalCount += delta;
}
Count-Min-Mean Sketch
对估计值取均值而非最小值，在计数较小时提供更准确的估计（但可能低估）。
复杂度分析
| 指标 | 复杂度 | 说明 |
|---|---|---|
| 空间复杂度 | O((1/ε)·ln(1/δ)) | 仅需存储二维计数矩阵 |
| 增加计数 | O(ln(1/δ)) | 与深度成正比 |
| 查询频率 | O(ln(1/δ)) | 取各哈希位置最小值 |
| 误差界 | ε·N | N为总元素数，概率至少1-δ |
以 ε=0.001, δ=0.001 为例：width ≈ 2719, depth ≈ 7，仅需约 2719×7×8 ≈ 152 KB 即可处理数十亿条记录。
应用场景
Count-Min Sketch在以下场景有广泛应用：
- 网络流量监控：实时统计IP地址、端口或URL的访问频率，检测DDoS攻击
- 热门商品统计：电商平台上实时追踪商品点击和购买热度
- 数据库查询优化：估计不同查询条件的出现频率，辅助索引选择
- 自然语言处理：统计词频分布，用于TF-IDF等特征计算
- 推荐系统：统计用户-物品交互频率，作为协同过滤的输入
与精确哈希表相比，CMS以约0.1%的精度损失换取了数百倍的空间压缩，是大数据流处理中不可或缺的概率工具。
