Flink SQL 压测最短闭环Print 验证正确性 + BlackHole 榨干性能上限（附 Join/Agg/TopN/UDF 模板）

1. 为什么要先 Print 再 BlackHole

很多人一上来就对着 ES/JDBC/S3 这类真实 Sink 压，得到的结果通常是“很慢 + 各种失败重试”，但你无法回答关键问题：

到底是 SQL 算得慢，还是写得慢？

Print 和 BlackHole 分别解决这两个维度：

• Print：把每条结果直接打到 task 日志里，肉眼可验、最适合查逻辑（但大流量会把日志打爆，性能极差）
• BlackHole：把输出吞掉，相当于 Linux 的/dev/null，最适合测算子链路吞吐上限（但它不验证外部写入正确性）

一句话：
先用 Print 把 SQL 的“语义”钉死，再用 BlackHole 把 SQL 的“性能上限”测出来。

2. 整体闭环长什么样

推荐你按这个顺序走：

1. 小流量 + Print：看结果是否符合预期（字段、Join 命中、Agg 口径、TopN 逻辑、RowKind）
2. 大流量 + BlackHole：看吞吐是否达标、是否背压、Checkpoint 是否拖慢
3. 真实 Sink 回归：再去调 ES/JDBC/FS 的 flush、batch、并发、失败策略

你会明显感觉：定位速度快很多，结论也更可靠。

3. 第一步：用 Print 做“正确性验收”（小流量）

3.1 创建 Print 表（SQL Sink）

CREATETABLEsink_print(user_idBIGINT,item_idBIGINT,cntBIGINT,window_endTIMESTAMP(3))WITH('connector'='print','print-identifier'='CHECK');

Print 输出会带 RowKind，例如：+I(...)、-U(...)、+U(...)，这在排查 Upsert/聚合更新时非常关键。

3.2 用 DataGen 快速造数据（可选）

没有 Kafka、没有真实数据也没关系，先用 DataGen 把链路跑通：

CREATETABLEgen_src(user_idBIGINT,item_idBIGINT,scoreINT,tsTIMESTAMP(3),WATERMARKFORtsASts-INTERVAL'2'SECOND)WITH('connector'='datagen','rows-per-second'='50','fields.user_id.min'='1','fields.user_id.max'='1000','fields.item_id.min'='1','fields.item_id.max'='100','fields.score.min'='0','fields.score.max'='100','fields.ts.max-past'='10 s');

rows-per-second 刻意调小，让你能看清楚每条输出和 RowKind。

3.3 把你的 SQL 塞进去（示例：窗口聚合）

INSERTINTOsink_printSELECTuser_id,item_id,COUNT(*)AScnt,window_endFROMTABLE(TUMBLE(TABLEgen_src,DESCRIPTOR(ts),INTERVAL'10'SECOND))GROUPBYuser_id,item_id,window_start,window_end;

正确性验收你应该检查什么：

• 口径：COUNT/ SUM 是否符合预期
• 时间：窗口边界是否正确、迟到数据是否影响结果
• RowKind：是否出现预期的更新（-U/+U）或撤回（-D）

如果这里没问题，再进入性能压测。

4. 第二步：把 Sink 换成 BlackHole，测“SQL 计算吞吐上限”（大流量）

4.1 创建 BlackHole 表

CREATETABLEsink_bh(user_idBIGINT,item_idBIGINT,cntBIGINT,window_endTIMESTAMP(3))WITH('connector'='blackhole');

4.2 同一段 SQL，只换 Sink

INSERTINTOsink_bhSELECTuser_id,item_id,COUNT(*)AScnt,window_endFROMTABLE(TUMBLE(TABLEgen_src,DESCRIPTOR(ts),INTERVAL'10'SECOND))GROUPBYuser_id,item_id,window_start,window_end;

4.3 把 DataGen 的速率提上去，逼出瓶颈

ALTERTABLEgen_srcSET('rows-per-second'='80000');

如果版本不支持 ALTER，就直接重新建表或改 DDL 重跑。

此时 BlackHole 会把外部写入成本完全消掉，你测到的基本就是：

• SQL 算子链吞吐
• shuffle/序列化/网络开销
• state/checkpoint 开销（如果有）

5. Join/Agg/TopN/UDF 压测模板（直接套用）

下面给你三类“最容易出性能问题”的 SQL 模板，你只要把字段名换成你自己的。

5.1 Join 模板（维表 Lookup / Regular Join）

维表 Lookup 常见瓶颈是外部访问或缓存策略，这种场景建议两段压测：

• 先把维表替换成“本地临时表/小表”测 join 算子开销
• 再接真实维表测外部依赖

Regular Join（流流 join）：

INSERTINTOsink_bhSELECTa.user_id,a.item_id,b.some_field,a.tsFROMstream_a aJOINstream_b bONa.user_id=b.user_idANDa.tsBETWEENb.ts-INTERVAL'5'SECONDANDb.ts+INTERVAL'5'SECOND;

重点观察：是否发生严重背压、是否某个 join key 倾斜。

5.2 聚合模板（Group Agg / Window Agg）

INSERTINTOsink_bhSELECTuser_id,COUNT(*)ASpv,SUM(score)ASscore_sumFROMgen_srcGROUPBYuser_id;

重点观察：state 增长速度、checkpoint 时长、RocksDB（如启用）压力。

5.3 TopN 模板（最容易背压）

INSERTINTOsink_bhSELECT*FROM(SELECTuser_id,item_id,score,ROW_NUMBER()OVER(PARTITIONBYuser_idORDERBYscoreDESC)ASrnFROMgen_src)WHERErn<=10;

TopN 常见问题：排序开销、状态膨胀、热点 user_id 倾斜。

5.4 UDF 模板（最容易 CPU 见底）

INSERTINTOsink_bhSELECTuser_id,my_udf(payload)ASxFROMgen_src;

建议你把 UDF 的逻辑拆成多段对比压测（只做解析 vs 解析+正则 vs 解析+网络访问），很快就能定位 CPU 黑洞。

6. 压测时重点看哪些指标，才能一眼判断瓶颈

只看“吞吐”是远远不够的。你至少要同时看这四类信号：

6.1 Backpressure（背压链路）

• 如果 BlackHole 也背压：瓶颈在计算、shuffle、序列化、状态、checkpoint
• 如果 BlackHole 不背压，但真实 Sink 背压：瓶颈在外部系统或 Sink 参数

6.2 吞吐与忙闲比例

• task busy 很高、吞吐上不去：CPU/算子重
• task busy 不高、吞吐上不去：可能是网络、序列化、锁竞争或 checkpoint 对齐

6.3 Checkpoint（尤其是 alignment）

• checkpoint duration 很长：state 大、写入慢或资源紧张
• alignment 时间异常：上下游并行度不匹配、数据倾斜导致 barrier 等待

6.4 GC 与内存

• Young GC 频繁：对象分配过多（UDF/JSON 解析/字符串拼接）
• Old GC/Full GC：内存压力大或状态/缓存设置不合理

你会发现：用这四类指标配合 Print/BlackHole 的分层压测，定位速度会比“盲调参数”快一个数量级。

7. 常见坑：别踩

1. Print 只能小流量
   大流量 Print 基本等于“用日志系统当消息队列”，吞吐会直接坍塌

2. BlackHole 只测上限，不代表真实 Sink 一定能达到
   真实 Sink 还涉及 bulk、batch、失败重试、限流、写入模型等

3. 倾斜是最隐蔽的性能杀手
   BlackHole 下仍然背压，很多时候不是算子复杂，而是热 key 把某个 subtask 打爆

4. Checkpoint 会显著影响压测结论
   建议至少做两组：关 checkpoint 看上限，开 checkpoint 看真实形态（更贴近生产）

8. 一份可复制的压测清单（拿去就用）

• 正确性阶段（Print，小流量）

  • RowKind 是否符合预期
  • Join 命中率、Agg 口径、TopN 结果是否正确
  • 水位线与窗口边界是否符合预期

• 性能阶段（BlackHole，大流量）

  • 提高 rows-per-second，找到吞吐拐点
  • 看背压是否出现，出现在哪个算子
  • 看 checkpoint duration 与 alignment
  • 看 GC 与 CPU 利用率

• 回归阶段（真实 Sink）

  • 再去调 flush/batch/并发/失败策略
  • 再看吞吐与延迟是否满足 SLA

9. 结语：把 SQL 贴出来，你就能得到“最短闭环”的定制方案

这套方法的精髓是：把不确定因素剥离掉，让每一步都只回答一个问题。

如果你把你要压测的那段 SQL（尤其是 join/agg/topn/UDF）贴出来，我可以按这篇文章的方法给你定制一套：

• 哪些地方先用 Print 验证
• DataGen 如何造数据更贴近你的 key 分布
• BlackHole 压测如何逐级加压找到极限
• 指标怎么看，才能把瓶颈钉到某个算子或某类开销