Apache DataSketches 快速入门与集成示例
Apache DataSketches 快速入门与集成实战，从环境准备、第一个 HLL 程序到流式大数据管道的去重计数、Top-K 与分位数聚合，附完整可运行代码、真实运行输出，以及 Spark 与 Flink 的集成思路。
Apache DataSketchesApache DataSketches2026/10/11
Apache DataSketches 快速入门与集成示例
Apache DataSketches 用固定大小的「草图（sketch）」在单次流式遍历中近似回答「有多少个不同值（基数）」「第 99 分位是多少」「出现最多的 Top-K 是谁」这类问题，内存与传输开销可预测、可合并。本文先给出可运行的最小入门程序，再给出一个分片本地聚合、汇总端合并的流式管道完整示例（含真实回显），最后给出 Spark / Flink 的集成思路。
一、快速入门
1.1 环境准备
在 Ubuntu 服务器上准备 JDK 与 Maven，并在 Maven 工程引入依赖：
apt-get install -y openjdk-11-jdk maven
<dependency>
<groupId>org.apache.datasketches</groupId>
<artifactId>datasketches-java</artifactId>
<version>6.1.1</version>
</dependency>
1.2 第一个程序：去重计数（HLL）
package com.example;
import org.apache.datasketches.hll.HllSketch;
import org.apache.datasketches.hll.TgtHllType;
public class FirstSketch {
public static void main(String[] args) {
HllSketch uv = new HllSketch(12, TgtHllType.HLL_8); // lgK=12
for (long i = 0; i < 1_000_000; i++) {
uv.update(i); // 100 万个不同用户
}
System.out.printf("UV 估计 = %.1f (真实 1000000)%n", uv.getEstimate());
System.out.printf("95%% 置信区间 = [%.1f, %.1f]%n",
uv.getLowerBound(2), uv.getUpperBound(2));
}
}
编译运行：
mvn -q -B compile
mvn -q -B dependency:build-classpath -Dmdep.outputFile=cp.txt
java -Xmx256m -Dfile.encoding=UTF-8 -cp target/classes:$(cat cp.txt) com.example.FirstSketch
实测输出：
UV 估计 = 990116.6 (真实 1000000)
95% 置信区间 = [964895.9, 1016101.2]
1.3 结果解读
- 100 万个不同值，HLL 用约 4KB（lgK=12）给出 990116.6 的估计，相对误差约 -0.99%；
getLowerBound(2)/getUpperBound(2)给出 ±2 个标准差区间；真实值 1000000 落在 [964895.9, 1016101.2] 内；- 相比用
HashSet精确去重的内存随基数线性增长，sketch 的内存只由lgK决定，与基数无关——这正是它在亿级流式去重里流行的原因。
二、核心概念速览
| 需求 | 选用 sketch | 关键参数 | 输出方法 |
|---|---|---|---|
| 基数（去重计数） | HLL / CPC | lgK | getEstimate() |
| 集合交并差 | Theta | lgK、seed | union / intersect / aNotB |
| 分位数 / 中位数 | KLL | k | getQuantiles(double[]) |
| 频繁项 / Top-K | Frequent Items | maxMapSize | getFrequentItems(ErrorType) |
所有 sketch 都支持 序列化→传输→反序列化 与 合并，这是分布式聚合的基础。
三、集成示例：流式管道的去重计数 + Top-K + 分位数
3.1 场景与设计
模拟一个实时点击流管道：8 个分片并行处理 100 万条事件，每条事件包含用户 ID、类别、延迟。每个分片本地维护三类 sketch：
- HLL：统计分片内去重用户数（UV）；
- Frequent Items：统计热门类别；
- KLL：统计延迟分位数。
汇总端只需接收各分片序列化后的 sketch 字节并合并，无需搬运明细数据：
- HLL 用
org.apache.datasketches.hll.Union.update(HllSketch)合并； - Frequent Items 用
merge(other)合并； - KLL 用
merge(other)合并。
3.2 完整代码
package com.example;
import org.apache.datasketches.hll.HllSketch;
import org.apache.datasketches.hll.TgtHllType;
import org.apache.datasketches.hll.Union;
import org.apache.datasketches.kll.KllFloatsSketch;
import org.apache.datasketches.frequencies.ItemsSketch;
import org.apache.datasketches.frequencies.ErrorType;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
import java.util.Set;
public class IntegrationDemo {
static final int SHARDS = 8;
static final int EVENTS = 1_000_000;
static final int DISTINCT_USERS = 300_000;
public static void main(String[] args) {
Random r = new Random(42);
HllSketch[] uv = new HllSketch[SHARDS];
ItemsSketch<String>[] cat = new ItemsSketch[SHARDS];
KllFloatsSketch[] lat = new KllFloatsSketch[SHARDS];
for (int s = 0; s < SHARDS; s++) {
uv[s] = new HllSketch(12, TgtHllType.HLL_8);
cat[s] = new ItemsSketch<>(64);
lat[s] = KllFloatsSketch.newHeapInstance(200);
}
Set<Integer> trueUsers = new HashSet<>();
Map<String, Integer> trueCat = new HashMap<>();
float[] allLat = new float[EVENTS];
String[] cats = {"c0", "c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9"};
double[] w = {40, 20, 12, 8, 6, 5, 4, 2, 2, 1};
for (int i = 0; i < EVENTS; i++) {
int user = r.nextInt(DISTINCT_USERS);
String category = pick(cats, w, r);
float latency = (float) (Math.abs(r.nextGaussian()) * 120 + 50);
int s = i % SHARDS;
uv[s].update(user);
cat[s].update(category);
lat[s].update(latency);
trueUsers.add(user);
trueCat.merge(category, 1, Integer::sum);
allLat[i] = latency;
}
Union uvUnion = new Union(12);
ItemsSketch<String> catAll = null;
KllFloatsSketch latAll = null;
for (int s = 0; s < SHARDS; s++) {
uvUnion.update(uv[s]);
if (latAll == null) { catAll = cat[s]; latAll = lat[s]; }
else { catAll.merge(cat[s]); latAll.merge(lat[s]); }
}
double uvEst = uvUnion.getResult().getEstimate();
System.out.printf("去重计数(UV): 真实=%d 估计=%.1f 误差=%.4f%%%n",
trueUsers.size(), uvEst, (uvEst - trueUsers.size()) / trueUsers.size() * 100);
System.out.println("Top-3 类别(Frequent Items):");
ItemsSketch.Row<String>[] rows = catAll.getFrequentItems(ErrorType.NO_FALSE_POSITIVES);
Arrays.sort(rows, (x, y) -> Long.compare(y.getEstimate(), x.getEstimate()));
for (int i = 0; i < 3; i++) {
String item = rows[i].getItem();
System.out.printf(" %s 估计=%d 真实=%d%n", item, rows[i].getEstimate(), trueCat.get(item));
}
Arrays.sort(allLat);
System.out.printf("延迟分位数(KLL): P50=%.1f(真%.1f) P95=%.1f(真%.1f) P99=%.1f(真%.1f)%n",
latAll.getQuantile(0.5), q(allLat, 0.50),
latAll.getQuantile(0.95), q(allLat, 0.95),
latAll.getQuantile(0.99), q(allLat, 0.99));
long localBytes = 0;
for (int s = 0; s < SHARDS; s++) {
localBytes += uv[s].toUpdatableByteArray().length;
localBytes += lat[s].toByteArray().length;
}
System.out.printf("8 分片本地 sketch 序列化总量 = %d 字节(约 %.1f KB)%n", localBytes, localBytes / 1024.0);
}
static String pick(String[] cats, double[] w, Random r) {
double tot = 0;
for (double x : w) tot += x;
double t = r.nextDouble() * tot;
double acc = 0;
for (int i = 0; i < cats.length; i++) {
acc += w[i];
if (t < acc) return cats[i];
}
return cats[cats.length - 1];
}
static float q(float[] sorted, double p) {
int idx = (int) Math.round(p * (sorted.length - 1));
return sorted[idx];
}
}
3.3 运行与输出
java -Xmx512m -Dfile.encoding=UTF-8 -cp target/classes:$(cat cp.txt) com.example.IntegrationDemo
实测输出：
去重计数(UV): 真实=289069 估计=288544.2 误差=-0.1815%
Top-3 类别(Frequent Items):
c0 估计=400291 真实=400291
c1 估计=199486 真实=199486
c2 估计=120273 真实=120273
延迟分位数(KLL): P50=131.1(真130.9) P95=283.1(真285.0) P99=362.6(真359.5)
8 分片本地 sketch 序列化总量 = 52176 字节(约 51.0 KB)
3.4 结果解读
- 去重计数：8 分片各自去重后合并，得到 288544.2，与真实 289069 的相对误差仅 -0.18%，说明 HLL 合并是无损的近似（误差只来自 sketch 本身，不因合并放大）；
- Top-K：频繁项计数与真实完全一致，
maxMapSize=64足以容纳 10 个类别； - 分位数：P50/P95/P99 与真实值相差均在 1～2ms 以内；
- 体积：100 万条明细若逐条传输需数十 MB，而 8 分片 sketch 合并总量仅约 51KB，跨网络传输开销降低数个百分点。
四、在 Spark 中集成（示例）
DataSketches 是普通 jar，可作为 Spark 的依赖（--jars 或打进 fat jar）。典型做法是每个分区本地构建 sketch，再 reduce 合并 sketch 字节：
// Java/Scala 思路示意（未在本次实测集群运行）
// RDD: mapPartitions 内构造 HLL，返回 byte[]
JavaRDD<byte[]> perPartition = events.mapPartitions(iter -> {
HllSketch sk = new HllSketch(12, TgtHllType.HLL_8);
iter.forEachRemaining(e -> sk.update(e.getUserId()));
return Collections.singletonList(sk.toCompactByteArray()).iterator();
});
// 汇总：把各分区字节反序列化后合并
byte[] merged = perPartition.reduce((a, b) -> {
Union u = new Union(12);
u.update(HllSketch.heapify(a));
u.update(HllSketch.heapify(b));
return u.getResult().toCompactByteArray();
});
double uv = HllSketch.heapify(merged).getEstimate();
要点：把「可合并的 sketch 字节」而非明细数据作为 shuffle 对象，可显著降低网络与内存。
五、在 Flink 中集成（示例）
在 Flink 中把 sketch 作为聚合累加器（AggregateFunction 的 ACC）最自然：
// 思路示意（未在本次实测集群运行）
public class HllAgg implements AggregateFunction<Event, HllSketch, Double> {
@Override public HllSketch createAccumulator() { return new HllSketch(12, TgtHllType.HLL_8); }
@Override public HllSketch add(Event e, HllSketch acc) { acc.update(e.getUserId()); return acc; }
@Override public Double getResult(HllSketch acc) { return acc.getEstimate(); }
@Override public HllSketch merge(HllSketch a, HllSketch b) {
Union u = new Union(12); u.update(a); u.update(b); return u.getResult();
}
}
配合 keyBy(...).window(...).aggregate(new HllAgg()) 即可在窗口内做去重计数；分位数与 Top-K 同理分别用 KLL 与 Frequent Items 作为累加器。
六、常见问题
Q1：估计值为什么不是整数？
sketch 是概率性近似，输出为浮点估计值；用 getLowerBound / getUpperBound 给出置信区间，而不是追求精确整数。
Q2：合并时报 seed 不一致怎么办？
确保参与同一次集合运算的所有 sketch 使用相同 seed。实测不同 seed 合并会抛 SketchesArgumentException: Incompatible Seed Hashes。
Q3：lgK / k 该取多大？
lgK 每 +1 误差约减半、体积约翻倍；百万级去重常用 lgK=12~14，KLL 分位数常用 k=200。按可接受误差反推即可。
Q4：可以在生产中长期持有 sketch 对象吗？
可以，但在跨进程传输时应对其序列化（toByteArray() / compact()），反序列化用 heapify / deserialize，并保证两端 datasketches-java 版本一致。
章节目录