📚 课程目录第3章实时数据接入链路设计

3. 实时数据接入链路设计

实时 OLAP 的查询延迟与数据新鲜度，几乎完全由接入链路决定。一条设计良好的接入链路，需要同时回答三个问题：数据从哪里来、以什么语义写入、写入后结构变了怎么办。本节围绕这三条主线展开，先讲清各类数据源的接入方式与适用边界，再落到 Exactly-Once 的实现机制，最后给出 Schema 演进的可执行策略。

一、数据源接入方式与选型

实时 OLAP 的上游通常分为三类：消息队列（Kafka、Pulsar）、流计算中间层（Flink）、数据库变更捕获（CDC）。它们不是互斥关系，而是常常串联成一条链路，例如 MySQL → Flink CDC → Kafka → Flink 清洗 → OLAP。

Kafka 直连接入是最常见的形态。OLAP 引擎（如 Doris Routine Load、StarRocks Routine Load、ClickHouse Kafka Engine）直接消费 Topic，省去中间计算层，延迟可压到秒级。适合“上游已经写入 Kafka、且数据基本干净”的场景。关键参数包括：desired_concurrent_number（消费并发）、max_batch_interval（微批间隔，默认 10s）、max_batch_rows（单批行数上限）。若上游是 JSON 且字段稳定，直连即可；若需要去重、维表关联、多流合并，则应引入 Flink。

Pulsar 接入与 Kafka 类似，但有两个差异值得注意：一是 Pulsar 支持多订阅模式（Exclusive / Shared / Failover / Key_Shared），Key_Shared 可保证同一 key 的顺序性，适合按用户 ID 聚合的场景；二是 Pulsar 的分层存储（Tiered Storage）让历史数据可自动卸载到对象存储，OLAP 侧消费积压时不必担心 Broker 磁盘打满。接入时通常使用 Pulsar Connector 或 Flink Pulsar Connector。

Flink 作为接入中间层是复杂链路的标配。Flink 承担 ETL、去重、维表 Lookup、多流 Join 后，再通过 Flink-Doris-Connector、Flink-StarRocks-Connector 或 Flink-ClickHouse-Connector 写入 OLAP。这种方式的优势是可以在写入前完成数据质量校验和字段映射，把“脏数据”挡在 OLAP 之外。

CDC 接入解决的是“数据库里的变更如何实时同步”。主流方案是 Debezium + Kafka，或 Flink CDC（基于 Debezium 引擎，支持全量+增量一体化）。CDC 输出的是变更事件，包含 op（c=insert, u=update, d=delete）、before、after 三部分。写入 OLAP 时，insert/update 通常转为 UPSERT，delete 需要根据 OLAP 是否支持删除来决定处理方式——Doris 的 Unique Key 模型支持按主键删除，ClickHouse 则常用 ALTER TABLE ... DELETE 或 ReplacingMergeTree + 标记位。

二、Exactly-Once 语义的落地机制

“Exactly-Once”在实时链路中不是一个开关，而是由三段协同保证的：源端可重放、计算端可快照、目标端可幂等。任何一段缺失，端到端就只能退化为 At-Least-Once。

源端可重放：Kafka 通过 offset 记录消费位置，Pulsar 通过 subscription cursor 记录。只要 offset/cursor 能被持久化并在故障后恢复，源端就具备重放能力。Flink 消费 Kafka 时，offset 会随 Checkpoint 一起写入状态后端。

计算端可快照：Flink 的 Checkpoint 机制是核心。开启 checkpointing 后，JobManager 周期性触发 Barrier 对齐，算子将状态快照写入 HDFS/S3，同时记录 Kafka offset。故障恢复时从最近一次成功的 Checkpoint 重启，状态和消费位置一起回滚，从而避免重复处理。关键配置示例：

env.enableCheckpointing(60000);          // 60s 一次
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(120000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

目标端可幂等：这是最容易被忽视的一环。即使 Flink 做到了 Exactly-Once，如果写入 OLAP 时没有去重机制，重放仍会产生重复行。Doris 的 Routine Load 通过 Label 机制实现：每个微批对应一个 Label，Doris 内部记录已导入的 Label，重复 Label 会被拒绝。StarRocks Stream Load 同理。ClickHouse 则依赖 ReplacingMergeTree 的 version 列或 FINAL 查询去重，属于最终一致而非严格 Exactly-Once。

判断链路是否真正 Exactly-Once，可以做一个简单验证：在 Flink 作业运行中手动 kill TaskManager，观察恢复后 OLAP 表中主键行数是否与源端一致。若出现重复行，说明目标端幂等未生效。

三、数据格式与 Schema 演进策略

接入链路中数据格式的选择直接影响解析开销和演进灵活性。常见格式对比如下：

格式	解析开销	Schema 演进	适用场景
JSON	中	灵活，字段可增删	日志、埋点、快速迭代
Avro	低	强 Schema，支持兼容性检查	Kafka + Schema Registry
Protobuf	低	强 Schema，向后兼容好	跨语言、高性能场景
CSV	低	弱，列顺序敏感	批量导入、简单场景

Schema 演进的核心原则是向后兼容优先：新增字段必须带默认值，删除字段先标记废弃而非直接移除，字段类型变更只允许“窄转宽”（如 int → long）。在 Kafka + Schema Registry 体系中，Avro 的兼容性级别可设为 BACKWARD（新 Schema 可读旧数据）或 FULL（双向兼容），生产环境建议至少 BACKWARD。

当上游表结构变更时，CDC 链路需要同步处理。Flink CDC 支持 schema.change 事件，可以配置为自动将 DDL 变更同步到下游。但 OLAP 侧的 Schema 变更通常需要手动执行 ALTER TABLE，因此建议在 Flink 作业中增加一个 DDL 拦截分支：检测到 ADD COLUMN 时，先调用 OLAP 的 DDL 接口加列，再继续写入数据；检测到 DROP COLUMN 时，仅在下游停止写入该字段，保留列以避免历史数据丢失。

Schema 演进中最危险的操作是“字段重命名”。在 Avro 中重命名需要设置别名（alias），否则旧数据无法解析；在 JSON 中重命名会导致下游字段映射失败。建议用“新增字段 + 双写过渡 + 下线旧字段”三步走，而不是直接改名。

最后给出一个可参考的接入链路检查清单，用于上线前逐项确认：

源端 offset/cursor 是否随 Checkpoint 持久化？
Flink Checkpoint 间隔是否小于 OLAP 微批间隔的 2 倍？
OLAP 写入是否启用 Label / 事务 ID 去重？
数据格式是否注册 Schema Registry 并设置兼容性级别？
CDC 的 delete 事件是否有明确的处理策略？
Schema 变更是否有 DDL 同步或人工审批流程？