管线韧性 · 生产必修¶

Explanation · 原理资深

一句话定位

数据管线上生产后 · 90% 的运维都在处理"数据不干净 / schema 变了 / sink 挂了 / 要回填"。这页集中讲这些横切主题——与具体 source / sink 无关 · 所有管线都会踩。

和其他页的边界

本页讲 "管线运行时的生产韧性"。区别于：

lakehouse/Schema Evolution · 讲湖表协议层如何演化（列用 ID）
ops/灾难恢复 · 讲湖表层灾备（表级 rollback / 跨区复制）
CDC 内核 · 讲 CDC 技术原理
事件时间 · Watermark · 讲流式时间语义

TL;DR

端到端 Exactly-once · Source 幂等 + Engine 2PC + Sink 事务三方握手缺一不可
Schema Evolution 传播 · Debezium → Schema Registry → Paimon / Iceberg 自动演化有边界
DLQ · 脏数据对策 · Poison pill 隔离 · 幂等重试 · 失败事件归档
Backfill 正确做法 · Flink savepoint 回退 + offset 重置 · 不是 Airflow 重跑流作业
Backpressure · 识别瓶颈段 + 针对性扩容
Cutover / Handoff · 生产最大事故面 · 4 类交接点（bulk→CDC · backlog 恢复 · DLQ 回放 · schema 穿多下游）
可观测性契约 · 8 个 SLI · 强制 vs 可选 · 自动降级触发条件

1. 端到端 Exactly-once · 三方握手¶

三方职责¶

Source (Kafka)    Engine (Flink)           Sink (Iceberg / Paimon)
─────────────     ─────────────            ──────────────────────
offset 可重放   → checkpoint barrier   →   pre-commit (写数据文件 · 但不切指针)
offset commit     ↓                        ↓
                  2PC coordinator         commit (CAS 切 current snapshot)
                  ↓
                  若 commit 失败:
                    Flink 从上一个 checkpoint 恢复重试
                    pre-commit 已写的数据文件成"孤儿"(对象存储里确实还在)
                    reader 看不到(未被 metadata 指针引用)
                    后续由 `remove_orphan_files` 清理

注：这里的"孤儿文件对象存储上仍存在 · reader 看不到"靠的是湖表的 metadata 指针语义（见 lakehouse/Snapshot）· 不是对象存储级别的隐藏。对象存储本身一旦 PUT 成功文件就是可见的。

三方都要满足条件才是真 exactly-once：

Source · 支持 offset 重放（Kafka 默认 · PG WAL 默认 · MySQL binlog 位点默认）
Engine · 实现两阶段提交（Flink 的 TwoPhaseCommitSinkFunction 家族）
Sink · 支持事务 commit · 或天然幂等（Iceberg 用 CAS 指针 · Paimon 用 snapshot id）

典型组合的 Exactly-once¶

组合	Exactly-once	说明
Kafka + Flink + Iceberg	✅	Flink IcebergSink + checkpoint + Kafka offset 协同
Kafka + Flink + Paimon	✅	Paimon 原生支持 2PC
Kafka + Spark Streaming + Iceberg	⚠️ 是但要正确配置	checkpointLocation + 幂等写
Kafka + Kafka Connect Iceberg Sink	⚠️	取决于 connector 版本 · 至少 at-least-once

"湖仓 Exactly-once" 的实际边界¶

你实际得到的：

Source 侧消息不重不漏（Kafka / binlog）
Sink 侧 commit 成功 = 数据已落盘
commit 失败回滚后：对象存储里会有临时写入的孤儿文件 + metadata 未指向——读者看不到（正常 OCC 语义）· GC 周期清

超出边界的事：

业务去重（同一主键多条合并）· 需要 Paimon primary-key merge 或 Iceberg MERGE INTO
跨表原子提交 · 单 commit 影响多表 · 需要 Nessie 或 Iceberg multi-table commit（详见 lake-table.md Multi-table Atomic Commit）

2. Schema Evolution 在管线传播¶

湖表协议层（Iceberg / Paimon）支持 schema 演化 · 但要让"DB ALTER TABLE"正确传到湖表需要一条完整链路：

flowchart LR
  src[(MySQL<br/>ALTER TABLE ADD COLUMN)] --> dbz[Debezium]
  dbz --> sr[Schema Registry<br/>Avro · JSON-Schema · Protobuf]
  sr --> sink[Flink Sink / Kafka Connect Sink]
  sink --> lake[(Paimon / Iceberg<br/>演化 schema)]

能自动传播的¶

变更类型	自动?	备注
Add column (nullable)	✅	湖表协议原生支持
Add column (NOT NULL with default)	✅	Iceberg v3（spec 已于 2025-06 正式 ratified）+ / Paimon 1.0（2025-01 GA）+
Drop column	✅	协议层支持 · 但下游读侧要处理 null
Rename column	⚠️	Iceberg 用 field_id 自动；Delta 需启用 column mapping
类型升级（`int → long`）	✅ 部分	Iceberg 允许兼容扩位

不自动传播的（需人工）¶

变更类型	手动原因
类型收窄（`long → int`）	需要数据兼容性验证
NOT NULL 约束变化	通常 fail · 先改 null 兼容再改约束
STRUCT 嵌套字段变化	部分 engine 不支持自动演化嵌套
Delta 未启用 column mapping 的 rename	下游看成 drop + add

Flink CDC 3.x 的 Schema Evolution Behavior¶

3.0+ Pipeline 声明式控制（放在 pipeline 块里）：

source:
  type: mysql
  # ...

sink:
  type: paimon
  # ...

pipeline:
  name: app-to-paimon
  parallelism: 4
  schema.change.behavior: try_evolve    # 取值: evolve / try_evolve / lenient / ignore / exception

各取值含义：

evolve · 遇到所有变更都尝试演化 · 不支持就 fail
try_evolve · 尝试演化 · 不支持告警 + 降级（日志上报 · 作业不崩）— 生产推荐
lenient · 宽松兼容模式 · 尽量演化 · 不 fail
ignore · 忽略 schema change · 继续跑（新字段会被丢弃）
exception · 遇到 schema change 直接 fail · 早期开发 / 严苛场景

3. DLQ · 脏数据对策¶

Poison Pill 问题¶

Poison pill · 某条消息格式异常导致下游反序列化 / 转换失败：

若 fail-fast → 整个 job 挂 · 后续积压
若 silent skip → 漏数据 · 问题被掩盖
正确 → DLQ（Dead-Letter Queue）+ 告警

三种策略¶

策略	何时用
Fail-fast	早期开发 · 不能容忍脏数据
Skip + 告警	日志类场景 · 可接受少量漏数据
DLQ（死信队列）	生产正确做法 · 脏消息写独立 topic / 对象存储目录 · 人工或下游修复

DLQ 实现 · Flink SideOutput¶

final OutputTag<String> dlqTag = new OutputTag<>("dlq"){};

SingleOutputStreamOperator<Order> mainStream = input
    .process(new ProcessFunction<String, Order>() {
        @Override
        public void processElement(String raw, Context ctx, Collector<Order> out) {
            try {
                out.collect(parse(raw));
            } catch (Exception e) {
                ctx.output(dlqTag, raw);  // 脏消息分流
            }
        }
    });

mainStream.getSideOutput(dlqTag)
    .addSink(dlqKafkaSink);  // 独立 topic · 供离线修复

关键：DLQ 建了要配审查机制——定期人工 / 自动报警脚本扫 DLQ topic 行数告警——否则脏消息积累到百万级没人看，等同没建。

4. Backfill · 回填与重放¶

什么场景要 Backfill¶

初次上线 · 历史数据补齐
修复 · 某段时间 pipeline 产出有 bug · 需要重跑
Schema 重大变化 · 全量重写修正

正确做法 · Flink Savepoint¶

# Step 1 · 停当前作业 · 保存 savepoint
bin/flink stop <jobid> \
    --savepointPath s3://savepoints/

# Step 2 · 修改逻辑后 · 从 savepoint 启动 · 重置 Kafka offset 到目标位置
bin/flink run -s s3://savepoints/savepoint-xxx \
    --allowNonRestoredState \
    -- \
    source.kafka.startup.mode=specific-offsets \
    source.kafka.startup.specific-offsets=topic:0:1234567,topic:1:2345678

错误做法¶

用 Airflow 回填流作业 · Airflow 是批调度器 · 回填产生并行 job → 双写（原流作业 + 回填作业）· 数据重复
直接改 Sink 表数据 · 绕过管线 · 破坏 lineage + exactly-once 语义
snapshot 过期了才回填 · 历史 snapshot 被 expire_snapshots 清了 · 不能 time travel 回放——回填要提前规划保留期

全量 + 增量合流¶

历史段（批处理）+ 增量段（流）合流要保证：

不重不漏 · 边界靠 timestamp / offset 明确
去重 · Iceberg MERGE INTO 或 Paimon partial-update
推荐 · Flink 流批一体模式（1.17+）· 一套代码双跑

5. Backpressure · 流控¶

识别信号¶

Flink UI Backpressure tab 显示每个 operator 的压力等级（OK / LOW / HIGH）：

Source → Parse → Transform → Sink
  HIGH    HIGH     HIGH       [瓶颈点]
  ← ← ← ← ← ← ← ← ← ←

背压通常从 sink 端传导回来——找"第一个变 OK"的 operator · 它上游就是瓶颈。

生产应对¶

症状	应对
Sink 慢（Iceberg commit 慢）	增 sink parallelism · 检查 commit 频率（太高 → 小文件 → 更慢）· 配合 Compaction
Transform 慢	拆子任务 · 加并行度 · 异步 IO
Source 快 Sink 慢	中间加 Kafka 缓冲 · 或降 source 速率
State 太大	换 RocksDB backend · 加 TTL · 检查 key 分布（防热点）
Shuffle 数据倾斜	加盐打散 key · 两阶段聚合

6. Cutover / Handoff · 管线最大的事故面¶

观察：生产中最容易翻车的不是某个组件坏掉 · 而是状态转换点——全量切增量 / Kafka backlog 恢复 / DLQ 回放 / schema 演进穿多下游。这一节把这些交接点立成一等主题。

四个关键 Handoff¶

Handoff 1 · Bulk → CDC 切换（模式 C 的核心难点）

Phase 1: Bulk 装载(全量 snapshot)
        ↓
Handoff: 对齐 source DB 的 binlog offset / WAL LSN
        ↓
Phase 2: CDC 增量(从 handoff offset 开始)

陷阱： - binlog 已回收——源 DB binlog retain 太短 · bulk 期间 offset 就被清 · 切换失败 - 重叠区处理不当——snapshot 和 binlog 有重叠（snapshot 读过的行 · binlog 里又有 update）· 不去重会双写 - 业务双写——切换期间 application 同时写老系统和新系统 · 数据不一致

正确做法： 1. 启动 bulk 前 · 确认源 DB binlog retain > 预估 bulk 时间 × 2 2. bulk 开始时立即记录 binlog offset / WAL LSN 3. bulk 完成后 · 从记录的 offset 开始 CDC · 不是当前最新 offset 4. 重叠区依赖 Flink CDC / Debezium 的 watermark 对齐去重 5. 业务切换窗口短一点 · 用 primary key + upsert 语义容忍瞬时双写

详见：Bulk Loading · 和流式入湖的配合 · CDC 内核 · snapshot+increment

Handoff 2 · 长时间停机后的 Kafka Backlog 恢复

作业 crash / 集群升级 / 机房切换 · 下游消费者停了几小时：

恢复后: offset-lag 爆表 · 下游要追赶
  → source 吞吐 + 下游处理能力能不能撑住?
  → checkpoint 间隔要不要临时调大?
  → sink 的 commit 频率要不要临时降?

策略： - 追赶期缩减 sink commit 频率（例如 1min → 10min）· 减少 metadata 冲击 - 关闭非必要 transform（例如调试日志 / side output） - 独立恢复作业 · 和主流作业并行不共资源 · 追上后切回

Handoff 3 · DLQ 回放（poison pill 修复后重新入主流）

DLQ 里的脏消息经人工修复后重新回放到主流：

DLQ 中脏消息 → 修复脚本 → 写入主 Kafka topic(或触发重跑)
                            ↓
                        保证幂等(主键 upsert · 或 unique event id)

陷阱： - 不幂等——重放产生主流的重复 - 时序错乱——DLQ 消息原本时间早 · 回放时当作"当前时间" · event-time 逻辑错 - 水位线后移——历史消息回放触发晚窗口重算 · 下游受冲击

正确做法： - 主流走 primary key upsert（Paimon PK 表 / Iceberg MERGE）保证幂等 - 回放消息保留原 event-time · 通过 side output 或 replay-only topic 走 - 下游预期"回放时段有窗口重算" · 不要在此期做关键决策

Handoff 4 · Schema Evolution 穿多下游

Debezium 检测到源 DB ALTER TABLE · 一条事件要穿透：

source DB → Debezium → Schema Registry → Kafka → Flink → Iceberg sink
                                                         → Paimon sink
                                                         → ES / 实时 sink

挑战：每个下游对 schema change 的支持度不同 · 一家 fail 就阻塞主流。

策略（见上 §2 Schema Evolution 传播）： - Flink CDC schema.change.behavior: try_evolve 让不兼容的下游降级而非 fail - 多 sink 场景优先考虑 Pipeline 能否按 sink 分叉（部分支持 · 部分跳过）

7. 可观测性契约 · 应该盯什么 SLI¶

观察：运维一条管线 · 光知道 "checkpoint 和重试" 不够——真正要回答的是 "上线后盯哪些指标才算知道它活着"。

必须监控的 8 个 SLI¶

SLI	描述	告警阈值参考
Source Lag	source 最新 offset - consumer 已消费 offset	流场景 > 1 min 告警
End-to-End Freshness	数据入库时间 - 事件原生成时间（event-time）	SLA 违约前 30% 告警
Commit 成功率	成功 commit 数 / 尝试 commit 数	< 99% 告警
Schema Error Rate	schema 解析失败 / 总消息数	> 0.1% 告警（pipeline 要人工介入）
DLQ Volume	DLQ topic 消息数增长速率	持续 > 0 且连续 N 分钟增长告警（有漏网脏数据）
Replay / Backfill Backlog	正在补跑的数据量（offset 范围）	手动任务 · 进度可视化而非阈值告警
Checkpoint Duration	单次 checkpoint 从 start 到 ack 的时长	> checkpoint interval 告警（表示追不上）
Sink Commit Latency	sink 从 pre-commit 到 commit 的耗时	P99 > 1min 告警（指针切换阻塞）

强制 vs 可选¶

强制监控（不配就不能上生产）： - Source Lag · Commit 成功率 · Schema Error Rate · DLQ Volume

强烈推荐： - End-to-End Freshness（SLA 的具体体现） - Checkpoint Duration

情境性： - Replay Backlog · 只在 backfill 时看 - Sink Commit Latency · 大 Iceberg 表 / 高频 commit 场景

自动降级触发条件¶

触发条件	自动响应
Source Lag 持续 > 5 min	扩 source parallelism · 告警 oncall
DLQ Volume > threshold 且持续增	暂停主流 · 人工分析 dirty message 样本
Schema Error Rate > 1%	暂停 · 不要带着错 schema 继续写湖表
Checkpoint Duration > 5 × interval	说明 state 膨胀或 sink 阻塞 · 告警
Commit 成功率 < 95%	sink 级问题 · 告警 + 查 catalog 冲突

和 ops/observability 的边界¶

本页 SLI 清单 · 针对 管线本身的运行时指标
ops/可观测性 · 讲 湖表层的可观测（表级 stats / query latency / 存储成本等）

两层都要盯 · 互为补充。

8. 陷阱¶

以为 Exactly-once 自动得到 · 三方都要配置 · 漏一环就降级 at-least-once
schema.change.behavior 设成 exception · 一改字段 pipeline 就崩 · 生产用 try_evolve
DLQ 建了不看 · 百万脏消息积累 · 等同没建
用 Airflow 跑流作业回填 · 双写事故常见来源
Backpressure 持续 HIGH 不管 · 延迟积压 / state 膨胀 / 磁盘告警连锁爆炸
回填没提前规划 snapshot 保留期 · 要回填时发现历史 snapshot 已 expire
Kafka retain 比 checkpoint 周期短 · checkpoint 还没成功 offset 已失效 · 恢复失败