离线T+1跑了一年，老板突然要实时大屏——怎么加实时链路不翻车？

// 改造前的数据接入路径
埋点数据：SDK → Kafka → MaxCompute(ODS) ← 已有Kafka ✅
业务数据：MySQL → DataWorks全量同步 → MaxCompute(ODS) ← 没有Kafka ❌
 
// 改造后的数据接入路径
埋点数据：SDK → Kafka → MaxCompute(ODS)  ← 不变
                    └──→ Flink(实时链路)  ← 新增复用
业务数据：MySQL → Canal → Kafka → MaxCompute(ODS)  ← 改用Kafka中转
                              └──→ Flink(实时链路)  ← 新增

-- 离线DWD（已有，跑在MaxCompute）
INSERT OVERWRITE TABLE dwd_ad_click_di PARTITION(dt='${bizdate}')
SELECT
    event_id,
    COALESCE(user_id, 'unknown') AS user_id,
    UPPER(channel) AS channel,
    country_code,
    TO_TIMESTAMP(click_time) AS click_ts
FROM ods_ad_click_log_di
WHERE dt = '${bizdate}' AND event_id IS NOT NULL;
 
-- 实时DWD（新建，跑在Flink）——口径必须完全一致！
INSERT INTO dwd_ad_click_stream  -- Kafka Topic
SELECT
    event_id,
    COALESCE(user_id, 'unknown') AS user_id,       -- 同样的空值处理
    UPPER(channel) AS channel,                       -- 同样的标准化
    country_code,
    TO_TIMESTAMP(FROM_UNIXTIME(click_time/1000)) AS click_ts
FROM ods_ad_click_kafka                             -- 源换成Kafka
WHERE event_id IS NOT NULL;                         -- 同样的过滤条件

注意对比上面两段SQL——逻辑完全一致，只是数据源从Hive表变成了Kafka Topic，时间处理方式略有不同（离线的click_time已经是字符串格式，实时的是毫秒时间戳）。

第三步：Flink实时DWS层 → 写入OLAP

实时DWS是最核心的一步——直接决定了大屏上看到的数据。我们第一个实时需求是'广告ROI实时大屏'，所以DWS层先做渠道级别的实时聚合：

-- 实时DWS：渠道ROI分钟级聚合
INSERT INTO dws_channel_roi_realtime  -- 写入Hologres
SELECT
    TUMBLE_START(click_ts, INTERVAL '1' MINUTE) AS stat_time,
    channel,
    country_code,
    COUNT(*) AS click_pv,
    COUNT(DISTINCT user_id) AS click_uv,
    SUM(CASE WHEN is_install = 1 THEN 1 ELSE 0 END) AS install_cnt,
    SUM(CASE WHEN is_paid = 1 THEN pay_amount ELSE 0 END) AS revenue
FROM dwd_ad_click_stream
GROUP BY
    TUMBLE(click_ts, INTERVAL '1' MINUTE),
    channel,
    country_code;

写入Hologres的策略：我们用的是Flink Hologres Connector的'insertOrUpdate'模式——相同主键的数据会覆盖更新。这样即使Flink作业重启导致数据重放，也不会产生重复数据。这就是幂等写入的价值。

第四步：对账验证——实时和离线必须对得上

实时链路上线后，最关键的一步是对账。我们建了一套自动对账机制：

-- 每日对账SQL：对比实时和离线的核心指标差异
SELECT
    r.channel,
    r.country_code,
    r.click_uv AS realtime_click_uv,
    o.click_uv AS offline_click_uv,
    -- 计算差异率
    ROUND(ABS(r.click_uv - o.click_uv) / NULLIF(o.click_uv, 0) * 100, 2)
        AS diff_pct,
    -- 差异>5%标红告警
    CASE WHEN ABS(r.click_uv - o.click_uv) / NULLIF(o.click_uv, 0) > 0.05
         THEN '⚠️ 差异过大' ELSE '✅ 正常' END AS status
FROM (
    -- 实时汇总（按天回收）
    SELECT channel, country_code, SUM(click_uv) AS click_uv
    FROM dws_channel_roi_realtime
    WHERE stat_time >= '${yesterday} 00:00:00'
      AND stat_time <  '${today} 00:00:00'
    GROUP BY channel, country_code
) r
FULL JOIN (
    -- 离线汇总
    SELECT channel, country_code, click_uv
    FROM dws_channel_roi_1d WHERE dt = '${yesterday}'
) o ON r.channel = o.channel AND r.country_code = o.country_code;

对账结果写入监控表，用Grafana展示每日差异趋势。我们的经验是：刚上线第一周差异率在8-10%，主要原因是实时链路的窗口边界问题（跨天数据归属不同）。调整Watermark策略后，差异率稳定在2-3%以内。

面试高频追问

追问一：改造过程中离线链路有没有受影响？

完全没有。这就是渐进式改造的核心优势——实时链路是在旁边'长'出来的，跟离线链路共享数据源（Kafka）但独立运行。唯一共享的是OLAP层的写入目标表——实时和离线写入Hologres的同一张表，但通过时间分区隔离：实时写当天的分钟级数据，离线写昨天的日级数据。离线写入时会覆盖实时的昨日数据，实现'离线修正实时'。

追问二：为什么不一开始就全部实时化？

三个原因：① 成本——实时Flink作业7×24小时常驻运行，资源消耗是离线的3-5倍，全部实时化预算扛不住；② 稳定性——实时链路的运维复杂度远高于离线，刚起步时团队经验不足，全面铺开风险太大；③ 必要性——80%的指标T+1就够了，没必要花5倍成本做成实时。先把最核心的1-2个需求跑通，验证稳定后再逐步扩展，这是最务实的做法。

追问三：如果实时链路挂了，业务方怎么办？

我们跟业务方提前沟通了降级预案：实时大屏挂了，自动切换到离线T+1数据展示（虽然不是最新的，但至少有数据）。在Hologres的查询层做了逻辑判断——如果实时表最近10分钟没有新数据，自动读离线表的数据展示。这个降级机制在实时链路升级或故障时起了大作用，业务方感知不到链路切换。

实战踩坑

改造过程中最大的坑不是技术，而是'口径不一致'。我在翻译离线DWD的SQL到Flink时，漏了一个COALESCE——离线SQL里user_id为空时填'unknown'，但我在Flink SQL里忘了加这个处理。结果实时链路计算的click_uv比离线少了约5%（因为NULL的user_id被COUNT DISTINCT排除了）。

这个bug排查了整整一天。一开始我怀疑是窗口边界问题、Watermark问题，对了半天都不对。最后逐个字段对比离线和实时的DWD SQL，才发现漏了COALESCE。教训是：翻译SQL时必须逐行逐字段对比，最好用diff工具辅助。

-- 踩坑：翻译遗漏导致口径不一致
 
-- 离线DWD（正确）
COALESCE(user_id, 'unknown') AS user_id  -- ✅ 空值填充
 
-- 实时DWD（漏了！）
user_id  -- ❌ NULL的user_id没处理
 
-- 影响：COUNT(DISTINCT user_id) 不计NULL
-- 离线：user_id=NULL的被填成'unknown'，会被COUNT
-- 实时：user_id=NULL的被忽略，UV偏少~5%
 
-- 修复后
COALESCE(user_id, 'unknown') AS user_id  -- ✅ 跟离线保持一致

防坑建议：建一个'口径对照表'，把离线和实时每个字段的处理逻辑放在一起对比。每新增一个实时指标，都走一遍对照检查。另外，对账机制是最后一道防线——只要对账在跑，口径不一致的问题就能在24小时内发现。

总结

在已有离线数仓上加实时链路，核心是'渐进式改造、四步闭环'——接入层Kafka化、实时DWD清洗、实时DWS聚合、对账验证。最重要的三个原则：① 离线链路不动，实时在旁边长出来；② 口径必须和离线完全一致，逐行对比SQL；③ 先做1-2个核心需求跑通闭环，再逐步扩展。面试时能讲清楚这套渐进式改造方案，会非常加分——因为大多数公司都是这种从离线到实时的演进路径，面试官听了会觉得你很有实战经验。

离线和实时的架构都搭好了，但最头疼的问题来了——实时和离线的数据对不上！下一篇聊口径一致性怎么保证，这是Lambda架构的终极挑战。

你们公司是从纯离线加的实时，还是一开始就设计了双链路？中间有没有踩过口径的坑？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

数仓项目实战面试通关手册

第6-10篇：口径治理与数据层设计

Wood聊大数据

9年大数据老兵，聊技术、聊职场、聊成长