Sink 幂等怎么验收

难度与前置

难度：进阶
前置：知道 source 重放 / checkpoint 恢复会导致重复发往 Sink；会查下游表唯一键。
读完能做什么：在 Sink 侧设计幂等键与写出语义，用故障注入与 SQL 验收「重试不脏数据」，而不是只背 EOS 名词。

适用场景

Flink / Spark Structured Streaming 写入 Kafka、JDBC、数仓表、HBase/Redis
作业 failover、Kafka 重投后下游出现重复行或重复累加
需要和业务约定：重复到达时覆盖、忽略还是合并
端到端 EOS 做不到或不划算时，用 Sink 幂等兜底

不适用：

专文展开两阶段提交 / Kafka 事务整链路（见 E2E EOS、Kafka EOS）
无主键、只追加且业务允许重复的审计日志（仍建议可追溯 msg_id）
纯批 INSERT OVERWRITE 分区幂等（见离线幂等）

真实场景（一条）

实时订单状态写 MySQL，Sink 用普通 INSERT。Checkpoint 恢复后同一 order_id 插入多行，客服系统状态乱跳。治理：下游唯一键 order_id；Sink 改为 UPSERT（ON DUPLICATE KEY UPDATE / 等价）；接受「最后一次写入获胜」并文档化版本字段。杀 TaskManager 重放后行数稳定，状态与订单域对账一致。

怎么做（步骤清单）

1. 先写清 Sink 幂等语义

text

对每个业务键 K：
- 同一 K 消息到达 N 次 → 下游只保留 1 条有效状态（覆盖）
- 或：保留多版本但查询只取最新（需 version/ts）
- 禁止：无键追加导致 SUM/COUNT 被放大

与整链路 EOS 的分工：

层次	解决什么
Source / 算子状态 + CP	计算不丢不重（边界见 EOS 文）
Sink 幂等	即使上游重投，下游也不脏
离线对账	发现漏网与漂移

2. 选幂等键（Idempotency Key）

Sink 类型	常用键	冲突策略
订单状态 JDBC	`order_id`	upsert 最新状态
事件明细	`event_id`	ignore / 主键冲突丢弃
Kafka 下游主题	key=`biz_id`，value 带 version	消费者再幂等
数仓分区表	`dt + 业务主键`	分区覆盖或 MERGE
KV（Redis）	key + 可选 lua 版本比较	set 或 CAS

sql

-- 下游表必须能表达唯一性
ALTER TABLE rt_order_status
  ADD PRIMARY KEY (order_id);
-- 或 UNIQUE KEY

3. JDBC / 维表类 Upsert 示例

sql

-- MySQL 示意
INSERT INTO rt_order_status (order_id, status, updated_at, src_ts)
VALUES (?, ?, NOW(), ?)
ON DUPLICATE KEY UPDATE
  status = IF(VALUES(src_ts) >= src_ts, VALUES(status), status),
  updated_at = IF(VALUES(src_ts) >= src_ts, VALUES(updated_at), updated_at),
  src_ts = GREATEST(src_ts, VALUES(src_ts));

要点：带 事件时间/版本，避免乱序覆盖新值（与乱序重复协同）。

4. Kafka Sink：key + 下游再防重

java

// 示意：以业务键为 Kafka key，便于分区内有序
producer.send(new ProducerRecord<>("rt-order", orderId, payload));

注意：

Kafka 日志本身可重；幂等在消费者或再下一跳 Sink
若用 Kafka 事务 EOS，仍建议下游有业务键兜底（边界清晰）

5. 数仓 / 湖表 Sink

sql

-- Iceberg/Hudi/JDBC MERGE 示意（引擎语法自替换）
MERGE INTO dwd_order_rt t
USING staging_order s
ON t.order_id = s.order_id
WHEN MATCHED AND s.src_ts >= t.src_ts THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

批式兜底：按 dt 覆盖写入 staging，再合并——与幂等写入同一思想。

6. 故障注入验收（必做）

正常跑出基线行数 / 校验和
杀掉 Sink 所在 TaskManager 或重启作业从 CP 恢复
对同一输入制造重复（重放一段 Kafka）
断言：下游主键无重复；度量值不翻倍

bash

echo "step,rows,distinct_pk,dup_pk,metric_sum"
# baseline,100000,100000,0,1.23e7
# after_failover_replay,100000,100000,0,1.23e7

7. 监控与告警

Sink 重试次数、失败队列（DLQ）长度
下游 dup_pk 日检 = 0
覆盖率：src_ts 回退写入次数（乱序）
与离线对账差（见对账）

8. 文档与责任边界

text

写进设计说明：
- 幂等键 = ?
- 冲突策略 = ignore / upsert-max-ts / sum-merge（慎用）
- 不保证什么（如：恰好一次对外副作用短信）
- 与 EOS 宣传口径如何对外解释

常见坑（至少牢记这些）

只开 Checkpoint 就宣称 Sink 幂等
外部系统无主键时，恢复照样插重复。
幂等键用会变的业务字段
如「手机号」变更导致双行；用稳定 order_id/event_id。
upsert 不带版本被乱序打旧
先到的新状态被后到的旧消息覆盖。
SUM 型指标用「再插入一笔」当更新
重放即翻倍；应状态表覆盖或可逆事件溯源。
DLQ 回放无幂等
毒消息修复重放二次污染。
和整篇 EOS 混谈，验收只测 happy path
Sink 验收必须含「杀进程 + 重放」。

如何验收（可量化）

[ ] 下游存在唯一键 / 等价约束或日检 SQL
[ ] failover + 重放后 dup_pk = 0，核心度量不翻倍
[ ] 乱序消息按版本策略行为符合文档
[ ] Sink 失败有重试上限与 DLQ，回放流程幂等
[ ] 设计文档写清与 EOS 的边界（不夸大）

sql

SELECT
  COUNT(*) AS rows,
  COUNT(DISTINCT order_id) AS pk_cnt,
  COUNT(*) - COUNT(DISTINCT order_id) AS dup_est
FROM rt_order_status;
-- 期望 dup_est = 0 且 rows = pk_cnt

sql

-- 度量不翻倍：与订单域快照对照
SELECT
  r.dt,
  r.gmv AS rt_gmv,
  o.gmv AS off_gmv,
  (r.gmv - o.gmv) * 1.0 / NULLIF(o.gmv, 0) AS rel_diff
FROM ads_gmv_rt r
JOIN ads_gmv_off o ON r.dt = o.dt
WHERE r.dt = '${biz_date}';

Sink 幂等怎么验收 ​

适用场景 ​

真实场景（一条） ​

怎么做（步骤清单） ​

1. 先写清 Sink 幂等语义 ​

2. 选幂等键（Idempotency Key） ​

3. JDBC / 维表类 Upsert 示例 ​

4. Kafka Sink：key + 下游再防重 ​

5. 数仓 / 湖表 Sink ​

6. 故障注入验收（必做） ​

7. 监控与告警 ​

8. 文档与责任边界 ​

常见坑（至少牢记这些） ​

如何验收（可量化） ​

相关面试题 ​