Spark SQL原生不支持HLL函数,approx_count_distinct精度不可控且不支持合并;需引入spark-hyperloglog等扩展,注册hll_init/hll_merge/hll_estimate三个UDAF实现可控精度的分阶段HLL估算。

Spark SQL 里没有内置 HLL 函数,得靠扩展
Spark SQL 原生不支持 HLL(HyperLogLog)聚合函数,approx_count_distinct 底层用的是类似算法但不可控精度、不支持合并。真要估算十亿级独立用户(比如跨天/跨分区去重),必须引入外部实现——最常用的是 spark-hyperloglog(由 aggregatefunction 提供)或基于 algebird 的 UDAF。直接写 SELECT HLL(user_id) 会报错:org.apache.spark.sql.AnalysisException: Undefined function: 'HLL'。
用 spark-hyperloglog 注册自定义聚合函数
先加依赖(Maven):com.github.shevek:spark-hyperloglog_2.12:1.0.0(注意 Scala 版本对齐 Spark)。然后在 SparkSession 初始化后注册:
import com.github.shevek.hyperloglog.HLLAggregator
spark.udf.register("hll_init", HLLAggregator.init _)
spark.udf.register("hll_merge", HLLAggregator.merge _)
spark.udf.register("hll_estimate", HLLAggregator.estimate _)
注意三点:
-
hll_init接受原始值(如user_id),返回 HLL sketch 字节数组(BinaryType) -
hll_merge用于合并多个 sketch,必须配合AGGREGATE或窗口函数使用,不能直接写在普通GROUP BY里 -
hll_estimate把 sketch 转成整数估计值,结果是近似值,误差率约 ±1.05%(默认精度)
实际估算:分阶段聚合避免 OOM
十亿级用户不能一次性跑 hll_init(user_id) 再 hll_merge——单个 task 会把所有 sketch 加载进内存,极易 GC 或溢出。正确做法是两层聚合:
- 第一层(按分区/天/地域等 key 分组):
SELECT dt, hll_init(user_id) AS sketch FROM logs GROUP BY dt - 第二层(全局合并):
SELECT hll_estimate(AGGREGATE(sketch, hll_init(), hll_merge)) AS uniq_users FROM daily_sketches
这样每 task 只处理本分区 sketch,合并阶段只传字节数组(通常 DISTRIBUTE BY rand() 打散数据,否则单个 reduce task 仍可能扛不住。
误差和存储代价比 approx_count_distinct 更可控
approx_count_distinct(user_id) 在十亿量级下误差可能达 ±5%,且无法保存中间 sketch 复用;而 HLL sketch 可序列化存储(比如存 Parquet 的 BinaryType 列),后续只需 hll_merge + hll_estimate 就能快速更新全量去重数。但要注意:
- sketch 对空值(
NULL)默认跳过,确保user_id已清洗 - 不同实现的 sketch 不兼容(比如 Druid 的 HLL 和
spark-hyperloglog不能混用) - 如果要用 Hive 表做下游消费,得把 sketch 转成 Base64 字符串再存,否则 Hive 不认 BinaryType
真正卡住的往往不是函数调用,而是 sketch 合并时的数据倾斜和序列化格式错配——别省那一步 EXPLAIN EXTENDED 看物理计划。

















