实时数仓与 CDC

本章目标：能讲清实时数仓的分层和典型链路，设计一条 MySQL CDC → Kafka → Flink → Doris 的分钟级 GMV 链路，并知道实时与离线怎么对账。 前置知识：读完 Kafka 入门、Flink 基础；了解离线数仓的 ODS/DWD/DWS/ADS。

1. 为什么需要它

离线数仓 T+1 产出，产品经理上午才能看到昨天的数据。实时数仓把延迟压到秒级或分钟级，服务三类需求：实时监控（大屏 GMV、支付成功率）、实时决策（库存预警、风控、推荐特征）、实时运营（当天活动效果）。

它不是“把 Hive SQL 跑快一点”，而是另一套链路：消息队列缓冲、流计算持续处理、OLAP 低延迟查询。它也不能完全替代离线：T+1 全量校准、复杂多表关联、历史回溯仍更适合离线。

2. 核心概念

2.1 Lambda、Kappa 与流批一体

Lambda：同一份数据走批处理和流处理两条路，批层保证准确，速度层补延迟，服务层合并结果。缺点是两套代码、两套口径，合并时容易不一致。

Kappa：只保留流处理，需要重算时从 Kafka 重放历史。优点是只有一套逻辑；缺点是对保留时间、重放成本和消息格式稳定性要求高。

流批一体：Flink 用同一套 API/SQL 处理有界和无界数据。它减少“实时一套、离线一套”的重复开发，不等于所有场景只留一套作业。

面试高频：不要背“Kappa 取代 Lambda”。实际常见的是“实时链路 + 离线校准”或“流批一体 + 湖仓”逐步演进，取决于成本和一致性要求。

2.2 实时数仓分层

目标和离线分层一样：ODS 保真、DWD 可追溯、DWS 可复用、ADS 面向查询。区别是实时层没有“凌晨全量重跑”，必须靠窗口、状态和 upsert 持续更新。

2.3 典型链路：MySQL CDC → Kafka → Flink → Doris

MySQL --binlog--> Debezium/Canal/Flink CDC --> Kafka(ODS)
                                                     ↓
                     Flink(清洗、去重、维表 join、窗口聚合)
                          ├──→ Kafka(DWD)
                          └──→ Doris(DWS/ADS)

CDC（Change Data Capture）把数据库变更日志变成事件流。Flink CDC 可直接读 MySQL binlog，先全量快照再增量；Debezium + Kafka Connect 生态成熟；Canal 在阿里系常用。选型看团队已有组件，不只看“哪个新”。

2.4 维表 join

普通双流 join 会无限保留两条流的状态，实时场景风险大。优先 lookup join 或预打宽。维表变化也要定义口径：商品改名后，历史订单算旧名还是新名？多数报表用最新维表，需要历史准确性时用版本表。

2.5 去重、upsert 与迟到数据

去重三种做法：按业务唯一键（如 event_id）去重；Flink SQL 用 ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY event_time) 取第一条；下游用唯一键 upsert，让重复写入覆盖。

upsert 是实时数仓的关键。Doris Unique Key、MySQL 唯一索引都能按主键覆盖；ClickHouse ReplacingMergeTree 只是最终去重，merge 时机不确定，查询要 FINAL 或按主键聚合，且只在同一分区内生效。选它时要接受“结果可能短暂重复”，不能当成强一致的实时点更新。

迟到数据分两级：水位线 + allowed lateness 在窗口内更新；超过容忍时间的走侧输出落迟到表，T+1 修正。如果 DWS 用 Doris Aggregate 模型只做 SUM，迟到修正很难减掉旧值；用 Unique Key 存窗口结果、重算整个窗口更稳。

2.6 实时指标怎么写

分钟级 GMV：按支付事件的 event_time 开 1 分钟滚动窗口，按品类聚合，订单去重。

INSERT INTO dws_trade_minute
SELECT TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start, category,
       COUNT(DISTINCT order_id) AS order_cnt, SUM(pay_amount) AS gmv
FROM pay_events WHERE event_type = 'pay'
GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE), category;

日活买家：一天内至少有一个订单事件的去重买家。没有登录和浏览数据时不要写 DAU，口径明确写成“日活买家”。全天 COUNT(DISTINCT buyer_id) 状态太大，常见做法是按分钟/小时输出买家明细，再在 Doris 用 BITMAP_UNION 或 COUNT(DISTINCT) 汇总。

SELECT dt, COUNT(DISTINCT buyer_id) AS active_buyers
FROM dwd_order_events
WHERE dt = '${biz_date}' GROUP BY dt;

订单状态转化率：用订单生命周期事件做漏斗，放在同一订单粒度上，避免把“买家数”和“订单数”混算。

SELECT dt,
  COUNT(DISTINCT CASE WHEN event_type = 'order_created' THEN order_id END) AS created_cnt,
  COUNT(DISTINCT CASE WHEN event_type = 'approved'      THEN order_id END) AS approved_cnt,
  COUNT(DISTINCT CASE WHEN event_type = 'shipped'       THEN order_id END) AS shipped_cnt,
  COUNT(DISTINCT CASE WHEN event_type = 'delivered'     THEN order_id END) AS delivered_cnt
FROM dwd_order_events
WHERE dt = '${biz_date}' GROUP BY dt;

转化率 = 下一步订单数 / 上一步订单数。必须写清事件定义和归因窗口（当天、7 天、会话内），否则和离线对不上。

2.7 离线与实时对账

差异是正常的，关键是能解释。常见原因：时间语义不同（事件时间 vs 分区日期）、迟到数据、重复消费、口径不同（含不含运费/退款/取消）、CDC 延迟、时区混用。

对账做法：同一天、同一口径、同一时间范围，分别汇总实时和离线 GMV，算差异率：

WITH rt AS (SELECT SUM(gmv) AS gmv FROM ads_gmv_realtime_daily WHERE dt = '${biz_date}'),
     off AS (SELECT SUM(gmv) AS gmv FROM ads_gmv_offline_daily WHERE dt = '${biz_date}')
SELECT rt.gmv AS realtime_gmv, off.gmv AS offline_gmv,
       ROUND((rt.gmv - off.gmv) / off.gmv, 4) AS diff_rate
FROM rt CROSS JOIN off;

超过阈值就告警。排查顺序：口径 → 时间范围 → 重复 → 迟到 → CDC 延迟 → 代码。实时链路通常以离线结果为准做 T+1 校准。

3. 动手示例

3.1 topic 设计

order_events  key = order_id  6 分区  保留 7 天
pay_events    key = order_id  6 分区  保留 7 天
review_events key = review_id 6 分区  保留 7 天
dim_product   key = product_id 3 分区 压缩清理（compact）

订单/支付按 order_id 保证同一订单有序；评价按 review_id 保证同一评价有序；维表 topic 开启 compact，只留每个 key 的最新值。

3.2 Flink SQL：支付事件 → 分钟 GMV → Doris

Kafka 消息字段为 order_id、category、pay_amount、event_type、pay_time，其中 pay_time 是 yyyy-MM-dd HH:mm:ss 字符串。Flink SQL：

CREATE TABLE pay_events (
  order_id STRING,
  category STRING,
  pay_amount DECIMAL(12, 2),
  event_type STRING,
  pay_time TIMESTAMP(3),
  WATERMARK FOR pay_time AS pay_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'pay_events',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'realtime_gmv',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);
CREATE TABLE doris_sink (
  window_start TIMESTAMP(3),
  category STRING,
  order_cnt BIGINT,
  gmv DECIMAL(18, 2)
) WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'table.identifier' = 'realtime.dws_trade_minute',
  'username' = 'root',
  'password' = '',
  'sink.label-prefix' = 'dws_trade_minute_job'
);
INSERT INTO doris_sink
SELECT
  TUMBLE_START(pay_time, INTERVAL '1' MINUTE) AS window_start,
  category,
  COUNT(DISTINCT order_id) AS order_cnt,
  SUM(pay_amount) AS gmv
FROM pay_events
WHERE event_type = 'pay'
GROUP BY TUMBLE(pay_time, INTERVAL '1' MINUTE), category;

Doris 结果表用 UNIQUE KEY(window_start, category)，保证同一窗口重放时覆盖而不是累加；生产副本数至少 3，本地练习可写 1。连接器的 label 前缀要按作业稳定配置，避免重复导入或标签冲突。

3.3 CDC 同步订单表

MySQL.orders --Flink CDC--> Kafka ods_orders_binlog --> Flink 清洗去重 --> Doris dwd_order

主键用 order_id，保证 insert/update/delete 落到同一行；删除事件不要直接丢，用软删除标记或连接器删除能力；全量快照和增量日志要衔接；开启 checkpoint，从 binlog 位点恢复。

4. 常见坑

1. 实时和离线口径不一致。一个含运费、一个不含；一个算支付、一个算有效订单。先统一指标定义。
2. 用处理时间做报表。重启或补数据后结果变化，无法和离线对齐。
3. DWS 用聚合模型承接可重放结果。重放一次 SUM 翻倍；用 Unique Key 存窗口结果。
4. 维表每条都查库。没有缓存，QPS 打满，反压传到 Kafka。
5. 双流 join 不设状态 TTL。状态无限增长，checkpoint 越来越大。
6. 迟到数据直接丢。实时少算，离线对不上；至少落迟到表。
7. CDC 只做增量、不做全量快照，且不看 Kafka 积压。历史数据缺失；源头积压时指标延迟，任务仍显示“正常”。

5. 在线练习

自测：

• 能设计订单流和用户行为流的 topic、key 和保留时间，并说明理由。
• 能说出实时数仓分层和离线数仓分层的对应关系。
• 能解释 CDC 同步中 update、delete 怎么落到 OLAP 表。
• 能说清端到端一致性和幂等写入的实现思路。

6. 面试延伸

面试题：Lambda 和 Kappa 怎么选？

是什么：Lambda 是批层 + 速度层双链路，Kappa 是只保留流处理、靠重放历史。

解决什么问题：Lambda 用批处理保证准确，Kappa 用一套代码降低维护和口径不一致。

我的实操边界：我做过离线数仓项目，理解实时分层和 CDC；没有在生产环境维护过 Lambda/Kappa 双链路，选型部分属于原理层面的了解。实际要落到成本和一致性：需要强校准就保留离线校准；消息可长期保留、流处理成熟才考虑 Kappa。很多公司是混合形态。

面试题：实时数仓和离线数仓的区别？

是什么：离线按天分区批量计算，实时按事件持续计算并低延迟输出。

解决什么问题：离线满足全量、复杂、可回溯；实时满足时效性。

我的实操边界：我离线项目跑通过 ODS 到 ADS 全链路，实时部分只做过本地 Kafka/Flink demo，没有上线过实时数仓。

区别分四点：时效、计算方式（批 vs 流）、存储（HDFS/Hive vs Kafka/OLAP）、更新方式（分区覆盖 vs upsert/窗口输出）。

面试题：实时链路怎么保证结果不重复？

是什么：靠 Kafka offset 与 Flink 状态一起 checkpoint，加上下游唯一键 upsert 或事务。

解决什么问题：故障恢复后重放数据，追加写会导致指标翻倍。

我的实操边界：我理解 checkpoint + Doris Unique Key 的组合，本地验证过 Kafka 生产和 Flink 窗口 demo；没有做过生产级故障注入。

答法：source 可重放；checkpoint 保存 offset 和窗口状态；sink 用唯一键覆盖或两阶段提交；最后用离线对账兜底。迟到数据见 Flink 进阶。

7. 本章验收

• 能画出 MySQL CDC → Kafka → Flink → Doris 的链路，说出每层存什么、算什么。
• 能比较 Lambda、Kappa 和流批一体，不把三者说成互相替代。
• 能设计订单/支付/评价 topic 的 key、分区和保留时间。
• 能写出分钟级 GMV 的 Flink SQL，并说明 Doris Unique Key 如何防止重放翻倍。
• 能说出维表 join 的三种方式和风险，列出实时与离线 GMV 差异的四个原因。
• 能解释实时日活买家为什么不能直接全天 COUNT(DISTINCT)，并说出一种替代方案。