近似计算详解

本篇介绍大数据场景下的近似计算技术，包括HyperLogLog去重、Bloom Filter判重、CountMin Sketch统计等，助你在精度与性能间找到平衡。

一、为什么需要近似计算

1.1 精确计算的痛点

1.2 近似计算优势

• 内存节省：通常只需KB级内存处理GB级数据
• 计算极速：时间复杂度从O(n)降至O(1)
• 误差可控：标准误差通常在1-5%范围内

二、HyperLogLog去重

2.1 原理概述

HyperLogLog（HLL）通过概率统计估算集合基数，核心思想：观察最大前导零数推断基数规模。

# Python简单实现原理
import hashlib

def hll_estimate(values, register_count=1024):
    registers = [0] * register_count
    for value in values:
        # 哈希
        hash_val = hashlib.md5(str(value).encode()).hexdigest()
        hash_int = int(hash_val, 16)

        # 获取前导零数
        x = bin(hash_int)[2:].zfill(128)
        leading_zeros = len(x) - len(x.lstrip('0'))

        # 获取索引位
        index = hash_int % register_count

        # 更新寄存器
        registers[index] = max(registers[index], leading_zeros)

    # 调和平均估算
    Z = sum(2 ** -r for r in registers)
    estimate = (register_count ** 2) / Z / 0.77351

    return int(estimate)

2.2 ClickHouse HyperLogLog

-- 精确COUNT DISTINCT
SELECT COUNT(DISTINCT user_id) FROM events;
-- 执行慢，内存占用大

-- 使用UNIQ精确去重
SELECT uniqExact(user_id) FROM events;

-- 使用uniqHLL近似去重（内存小，速度快）
SELECT uniqHLL(user_id) FROM events;

-- 查看估算误差
SELECT
    uniqHLL(user_id) AS approx_distinct,
    uniqExact(user_id) AS exact_distinct
FROM events;

-- 多列组合去重
SELECT uniqHLL(user_id, device_id) FROM events;

-- 带条件去重
SELECT
    uniqHLLIf(user_id, event_type = 'click') AS click_users,
    ununiqHLLIf(user_id, event_type = 'view') AS view_users
FROM events;

2.3 Hive HyperLogLog

Hive通过approx_count_distinct或HLL扩展实现。

-- 方法1：内置近似函数
SELECT approx_count_distinct(user_id) FROM events;
-- 误差约2%

-- 方法2：使用HLL扩展（需要启用）
SET hive.groupby.estimators.enabled=true;

-- 方法3：手动实现HLL
ADD JAR /path/to/hll-udf.jar;

CREATE TEMPORARY FUNCTION hll_count AS 'com.example.HyperLogLogUDF';

SELECT hll_count(user_id) FROM events;

2.4 MaxCompute HyperLogLog

-- 使用HyperLogLog函数（需开启配置）
SET odps.sql.allow.fullscan=true;

-- 近似COUNT DISTINCT
SELECT
    GET_IDCARD_NO_HASH(user_id) AS hash_id,
    -- 结合其他近似函数
    COUNT(DISTINCT CASE WHEN dt = '2024-01-01' THEN user_id END) AS daily_uv
FROM events;

2.5 误差对比

-- 实际对比测试
SELECT
    'uniqExact' AS method,
    uniqExact(user_id) AS result
FROM events
UNION ALL
SELECT
    'uniqHLL',
    uniqHLL(user_id)
FROM events;

三、Bloom Filter判重

3.1 原理概述

Bloom Filter使用位数组和多个哈希函数，特点：

• 空间极省：每位可表示是否存在
• 仅支持添加和查询，不支持删除
• 存在假阳性（可能误判存在），不存在假阴性

# Python简单实现
import math
import hashlib

class BloomFilter:
    def __init__(self, size, hash_count):
        self.size = size
        self.hash_count = hash_count
        self.bit_array = [0] * size

    def add(self, item):
        for seed in range(self.hash_count):
            hash_val = int(hashlib.md5(f"{item}{seed}".encode()).hexdigest(), 16)
            index = hash_val % self.size
            self.bit_array[index] = 1

    def contains(self, item):
        for seed in range(self.hash_count):
            hash_val = int(hashlib.md5(f"{item}{seed}".encode()).hexdigest(), 16)
            index = hash_val % self.size
            if self.bit_array[index] == 0:
                return False
        return True

3.2 ClickHouse Bloom Filter

-- 创建Bloom Filter索引
CREATE TABLE users (
    user_id UInt64,
    email String,
    INDEX bf_email(email, 2048) TYPE bloom_filter GRANULARITY 1
) ENGINE = MergeTree()
ORDER BY user_id;

-- 使用bloom_filter函数
SELECT
    user_id,
    email,
    bloomFilterTest('email_hashes', email) AS may_exist
FROM users;

-- 手动创建Bloom Filter
SELECT
    bloomFilterHash('test@example.com') AS hash;

-- 批量检查重复
SELECT
    hasDuplicate(
        arrayMap(x -> hash(x), arrayCol),
        bloomFilterParameters(1000000, 3)
    ) AS has_dups;

3.3 Hive Bloom Filter

ADD JAR /path/to/bloom-filter-udf.jar;

CREATE TEMPORARY FUNCTION bloom_filter AS 'com.example.BloomFilterUDF';

-- 创建Bloom Filter
SELECT bloom_filter('init', 1000000) AS bf FROM dummy;

-- 添加元素
SELECT bloom_filter('add', bf, 'user1') FROM table;

-- 检查是否存在
SELECT bloom_filter('test', bf, 'user1') FROM table;

3.4 实际应用场景

-- 场景1：实时UV去重（结合Kafka）
-- 生产端：发送数据时带HLL摘要
INSERT INTO sink_stream
SELECT
    user_id,
    hyperLogLogBuild(user_id) AS hll_state
FROM source_stream
GROUP BY window_start, window_end;

-- 消费端：合并HLL并计算
SELECT
    window,
    hyperLogLogMerge(hll_state) AS uv
FROM sink_stream
GROUP BY window;

-- 场景2：用户标签去重合并
SELECT
    user_id,
    collect_set(tag) AS all_tags,
    size(collect_set(tag)) AS tag_count
FROM user_tags
GROUP BY user_id;

四、CountMin Sketch统计

4.1 原理概述

CountMin Sketch用于估算元素出现频率，类似于Bloom Filter但存储计数。

结构：d个哈希函数 × w个计数器
查询频率 = min{counter[i][h_i(x)] for i in 1..d}

4.2 ClickHouse实现

-- 使用windowFunnel统计事件序列
SELECT
    user_id,
    windowFunnel(3600)(event_time,
        event = 'view',
        event = 'add_cart',
        event = 'checkout',
        event = 'pay'
    ) AS funnel_level
FROM user_events
GROUP BY user_id;

-- 统计Top K元素
SELECT
    topK(10)(product_id) AS popular_products
FROM product_views;

4.3 Hive CountMin Sketch

ADD JAR /path-to/cmsketch-udf.jar;

CREATE TEMPORARY FUNCTION cms_estimate AS 'com.example.CountMinSketchUDF';

-- 初始化Sketch
SELECT cms_init(0.001, 0.01) AS sketch FROM dual;

-- 添加计数
SELECT cms_add(sketch, 'item1', 1) FROM table;

-- 查询频率
SELECT cms_estimate(sketch, 'item1') FROM table;

-- 多用户频率统计
SELECT
    user_id,
    cms_estimate(cms_sketch, item_id) AS freq
FROM user_items
GROUP BY user_id, item_id;

五、常用近似函数速查

5.1 ClickHouse

-- 去重计数
uniqHLL(expr)                    -- HyperLogLog估算
uniqHLLIf(expr, condition)      -- 带条件
uniqExact(expr)                 -- 精确去重

-- 频率统计
topK(k)(expr)                   -- Top K高频值
quantileDeterministic(x, determinator) -- 分位数

-- 聚合状态
uniqHLLState(expr)              -- 获取聚合状态
uniqHLLMerge(state)             -- 合并多个状态

5.2 Hive

-- 近似聚合
approx_count_distinct(expr)      -- 近似去重
approx_percentile(col, percentage) -- 近似分位数

-- 集合操作
collect_set(col)                 -- 收集不重复值
collect_list(col)                -- 收集所有值

5.3 误差控制

-- ClickHouse设置HLL精度
SET max_execution_time = 60;
SET max_memory_usage = 8000000000;

-- Hive设置估算精度
SET hive.groupby.estimators.error.rate=0.02;
SET hive.map.aggr.hash.min.reduction=0.5;

六、实战：实时UV统计系统

6.1 架构设计

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  数据采集   │ -> │  Kafka队列  │ -> │  Flink计算  │ -> │  结果存储   │
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘
                                        │
                                        v
                               ┌─────────────────┐
                               │  HyperLogLog    │
                               │  状态存储       │
                               └─────────────────┘

6.2 实现代码

-- Flink SQL实现
CREATE TABLE source (
    user_id STRING,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_events',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
);

CREATE TABLE sink (
    window_start TIMESTAMP,
    window_end TIMESTAMP,
    uv BIGINT,
    update_time TIMESTAMP
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/analytics',
    'table-name' = 'uv_stats'
);

-- 使用HLL进行UV统计
INSERT INTO sink
SELECT
    window_start,
    window_end,
    CARDINALITY(HLL(user_id)) AS uv,  -- 或使用内置HLL函数
    CURRENT_TIMESTAMP AS update_time
FROM TUMBLE(source.event_time, INTERVAL '1' HOUR)
GROUP BY window_start, window_end;

6.3 结果对比

七、精度调优指南

7.1 内存与精度权衡

-- ClickHouse HLL精度调整
-- 寄存器数量越多，精度越高
-- 16384寄存器 ≈ 0.5%误差
-- 4096寄存器 ≈ 1%误差
-- 1024寄存器 ≈ 2%误差

SELECT uniqHLL(user_id) FROM events;  -- 默认精度

-- 自定义精度（如果支持）
SELECT uniqHLL(user_id, 16384) FROM events;

7.2 选择合适的算法

7.3 常见问题

-- Q：HLL结果比实际值大？
-- A：正常现象，HLL存在约1-5%的正向偏差

-- Q：能否提高精度？
-- A：增加寄存器数量，但内存开销增大

-- Q：多个数据源如何合并？
-- A：使用HLL State + Merge
SELECT uniqHLLMerge(hll_state)
FROM (
    SELECT uniqHLLState(user_id) AS hll_state
    FROM table1
    UNION ALL
    SELECT uniqHLLState(user_id) AS hll_state
    FROM table2
) combined;

八、总结

近似计算的核心思想是：用可控的误差换取极大的性能和空间优势。在大多数业务场景下，1-5%的误差完全可以接受。