一、背景与痛点

1. 凌晨 2 点 DataArts 调度跑 ETL
2. 抽数到 OBS 落 Parquet
3. Hive 跑汇总 SQL，约 40 分钟
4. 早上 8 点财务打开 Excel 报表

• T+1 太慢：大促当天老板问“现在卖了多少”，没人答得上来
• 门店调价看不到效果：上午调价，下午才知道有没有用
• 库存预警滞后：热销 SKU 断货两天才发现
• 看板靠 Excel：数据口径每次都不一样，财务和业务打架
• 扩容麻烦：月底结账那几天 Hive 队列堵死

二、方案设计

2.1 为什么选 DLI + DLV

结论：查询并发不高但要快、要流式、要省钱的场景，DLI 的按量付费 + 原生流批一体最合适。如果每天要跑 PB 级批处理，还是回 DataArts + MRS。

2.2 整体架构

业务侧
├── 200 家门店 POS → RDS MySQL（订单库）
├── 小程序订单     → RDS MySQL
└── 库存系统       → RDS MySQL
        │
        ▼ CDC 实时捕获
数据接入
├── DataArts Studio CDM（或 Kafka + Debezium）
└── 写入 Kafka topic: retail_cdc
        │
        ▼
实时计算（DLI）
├── 源表：Kafka 流表 orders_stream
├── 维表：RDS 商品维 / 门店维（DLI 跨源）
├── 流作业 1：1 分钟窗口聚合 GMV / 订单量
├── 流作业 2：SKU 销量 TopN
└── 结果写入 DLI 表 dws_realtime
        │
        ▼
可视化（DLV）
├── 实时 GMV 大屏
├── 门店热力地图
├── SKU 爆款榜
└── 库存预警列表

2.3 分层建模

• ODS：Kafka 流表，原始 CDC 事件
• DWD：明细宽表，清洗去重后的订单粒度
• DWS：1 分钟窗口聚合结果（GMV、订单数、客单价）
• ADS：DLV 直接查的应用表，字段命名面向看板

三、实操步骤

3.1 打通 CDC 采集

1. DataArts 控制台 → 数据集成 CDM → 新建集群（1 个 micro 规格够用）
2. 新建连接：
   • 源端：RDS MySQL（填 VPC 内地址、账号、开启 binlog 的库）
   • 目的端：Kafka（华为云 DMS Kafka 实例）
3. 新建作业：实时同步 → 选表 t_order、t_order_item、t_sku
4. 同步类型勾 增量（CDC），起始位点选“当前时间”
5. 目标 topic 命名为 retail_cdc，分区数 12（按门店 ID hash）

避坑：MySQL 必须开 binlog_format=ROW，且同步账号要有 REPLICATION SLAVE, REPLICATION CLIENT 权限，否则 CDM 起不来增量。

3.2 DLI 建库建表

CREATE DATABASE IF NOT EXISTS retail_dw
WITH (LOCATION = 'obs://retail-dw/')
COMMENT '零售实时数仓';

CREATE TABLE IF NOT EXISTS retail_dw.ods_orders_stream (
    order_id      STRING,
    store_id      INT,
    sku_id        STRING,
    qty           INT,
    amount        DECIMAL(18,2),
    pay_time      TIMESTAMP,
    op_type       STRING,   -- c/u/d 来自 CDC
    event_time    TIMESTAMP,
    WATERMARK FOR event_time AS event_time - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'kafka.bootstrap.servers' = '10.0.0.11:9092,10.0.0.12:9092',
    'topic' = 'retail_cdc',
    'kafka.group.id' = 'dli-retail-ods',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);

CREATE TABLE IF NOT EXISTS retail_dw.dim_sku (
    sku_id    STRING,
    sku_name  STRING,
    category  STRING,
    price     DECIMAL(18,2)
) WITH (
    'connector' = 'rds',
    'url' = 'jdbc:mysql://10.0.1.20:3306/retail',
    'table_name' = 't_sku',
    'user_name' = '${secret:rds_user}',
    'password' = '${secret:rds_pwd}'
);

账号密码走 DLI 的凭证管理 ${secret:xxx}，别硬编码在 SQL 里，审计过不了。

CREATE TABLE IF NOT EXISTS retail_dw.dws_gmv_1min (
    window_start  TIMESTAMP,
    window_end    TIMESTAMP,
    store_id      INT,
    order_cnt     BIGINT,
    gmv           DECIMAL(18,2)
) WITH (
    'connector' = 'dli',
    'format' = 'parquet'
);

3.3 提交流式作业

INSERT INTO retail_dw.dws_gmv_1min
SELECT
    TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(event_time, INTERVAL '1' MINUTE)   AS window_end,
    store_id,
    COUNT(DISTINCT order_id) AS order_cnt,
    SUM(amount)              AS gmv
FROM retail_dw.ods_orders_stream
WHERE op_type IN ('c','u')          -- 过滤删除事件
GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE), store_id;

• 引擎：Flink 1.15
• CU 数：4（起步，压测后再调）
• 并行度：4
• Checkpoint：60 秒，存 OBS
• 失败重启策略：fixed-delay，3 次

INSERT INTO retail_dw.ads_hot_sku
SELECT window_end, sku_id, sku_name, SUM(qty) AS qty
FROM (
    SELECT
        TUMBLE_END(o.event_time, INTERVAL '10' MINUTE) AS window_end,
        o.sku_id, d.sku_name, SUM(o.qty) AS qty,
        ROW_NUMBER() OVER (
            PARTITION BY TUMBLE(o.event_time, INTERVAL '10' MINUTE)
            ORDER BY SUM(o.qty) DESC
        ) AS rn
    FROM retail_dw.ods_orders_stream o
    JOIN retail_dw.dim_sku FOR SYSTEM_TIME AS OF o.event_time AS d
      ON o.sku_id = d.sku_id
    GROUP BY TUMBLE(o.event_time, INTERVAL '10' MINUTE), o.sku_id, d.sku_name
) WHERE rn <= 20;

3.4 DLV 建大屏

1. DLV 控制台 → 新建大屏 retail-realtime-board
2. 添加数据源 → 选 DLI，填 Region、库 retail_dw、AK/SK（用委托授权更安全）
3. 拖组件：

1. 全局滤镜加“门店下拉框”，联动所有图表
2. 自动刷新设 30 秒
3. 发布 → 生成分享链接，投到会议室大屏

技巧：DLV 的 SQL 数据集支持变量替换，${store_filter} 接下拉框，比建 N 个数据集清爽得多。

3.5 权限与审计

• DLI 库表权限用 IAM 细粒度：看板账号只给 SELECT on retail_dw
• DLV 数据源用委托授权，不要在前端存 AK/SK
• 开 DLI 审计日志落 LTS，谁在什么时候查了什么留痕
• 大屏分享链接加 token 有效期，外部合作方看完自动失效

四、验证结果

五、生产避坑清单

1. Watermark 设太小：设 5 秒，网络一抖迟到数据全丢，GMV 对不上账 → 按业务迟到分布设 30-60 秒
2. 维表不做 temporal join：直接用普通 join，商品改名后历史数据全变新名字 → 用 FOR SYSTEM_TIME AS OF
3. 流作业不设 Checkpoint：Flink 挂了从零开始重放，Kafka 积压炸掉 → 必开，60 秒一次落 OBS
4. DLV 数据集写 SELECT *：字段一变大屏全崩 → 显式列名
5. AK/SK 硬编码在 SQL 或前端：安全扫描直接红灯 → 走委托或凭证管理
6. Kafka 分区数拍脑袋：分区少于 Flink 并行度，部分并行度空转 → 分区数 ≥ 并行度 × 2
7. 不做幂等：CDC 重放导致 GMV 翻倍 → 用 order_id + op_type 做去重或写 upsert Kafka
8. 大屏自动刷新 5 秒：DLI 被自己看板打挂 → 30 秒足够，热点数据加 DLV 缓存
9. 删除事件不过滤：op_type='d' 的记录也进了 SUM，GMV 变负 → WHERE 里显式排除
10. 跨 Region 拉维表：DLI 在北京、RDS 在上海，每条维表查询 30ms，吞吐直接腰斩 → 同 Region 同 VPC

• Watermark = P99 迟到时长，宁可慢一点也别丢数
• 流作业 CU 起步给够，压测后再降，别一上来 1CU 卡到怀疑人生
• DWS 表按天分区，历史数据自动落冷存储省 OBS 钱
• 大屏所有 SQL 走 ADS 层应用表，不让前端直接查 DWD 明细
• 每晚跑一次对账作业：实时累计 vs T+1 离线，差异 > 0.5% 告警
• 流作业加监控：Kafka lag、Checkpoint 时长、反压状态三指标进云监控

六、成本结构梳理

对比原方案：Hive 集群 8 台常驻 ECS 月费 ≈ 1.9 万，DLI 按量 + Kafka + CDM 合计 ≈ 8.5 千，降本 55%，还顺带把时延从 T+1 干到分钟级。

七、总结与扩展

• 门店调价 → 15 分钟内看到销量曲线拐点
• 爆款断货 → 库存预警比原来早 1.5 天
• 大促指挥 → 老板自己拖大屏，不用等分析师出表
• 数据口径 → 统一在 DWS 层，财务和业务不再各算各的

1. 流式不等于全流：明细走流、汇总走流、复杂的同环比还是留 T+1 批，混着用最省
2. 看板性能瓶颈在 SQL 不在前端：ADS 层把该聚合的聚合完，DLV 只做渲染
3. 成本杀手是常驻 CU：查询型作业一定走按量队列，别跟流作业抢固定资源

1. 异常检测：DLI 流作业里接 UDF，GMV 同比跌 30% 自动推企微告警
2. 预测看板：ModelArts 训练销量预测模型，DLV 上叠“今日预估”虚线
3. 门店自助分析：给店长开 DLV 编辑权限，只看自己门店分片数据
4. 批流一体回溯：需要重算历史时，同一份 SQL 切 batch 模式跑 OBS 历史分区
5. 移动端适配：DLV 大屏另出一版手机布局，区域经理路上也能刷

本文为原创实时数仓实践，基于华为云 DLI、DLV、DMS Kafka、CDM 整理，操作步骤以当前控制台页面为准。文中时延与成本数据为特定环境实测，实际表现随数据量与规格浮动。转载请注明出处。