【Flink 30天】Day24-27 实时数仓规范：ODS→DWD→DWS→ADS 分层 + Temporal Join + 生产最佳实践

Day 24-27：实时数仓规范 & 生产最佳实践

1. 实时数仓分层

Kafka（ODS 原始数据）
↓ Flink 清洗、过滤、格式标准化
DWD（数据明细层 / Kafka Topic）
↓ Flink 窗口聚合、Join 维度
DWS（数据汇总层 / Kafka Topic 或 ClickHouse）
↓ 简单查询/汇总
ADS（应用数据层 / ClickHouse / Redis / MySQL）
↓
业务系统、大屏、报表

2. DWD 层：实时宽表设计

核心思想：通过 Flink Join 将多个 Kafka topic 的数据拼宽，形成一张包含所有维度的宽表。

— 订单事件 + 用户维度 + 商品维度 → 订单宽表
INSERT INTO dwd_order_wide
SELECT
o.order_id,
o.user_id,
u.username,
u.city,
o.product_id,
p.product_name,
p.category,
o.amount,
o.ts
FROM ods_order o
— Lookup Join：实时查维度表（推荐！不存状态）
JOIN dim_user FOR SYSTEM_TIME AS OF o.ts AS u ON o.user_id = u.user_id
JOIN dim_product FOR SYSTEM_TIME AS OF o.ts AS p ON o.product_id = p.product_id;

3. 维度表关联：Temporal Join（时态 Join）

Temporal Join = 用流数据的事件时间，关联维度表在那个时刻的快照。

适用于：维度数据会变化（如商品价格），需要关联历史时刻的值。

— 维度表（MySQL CDC 同步到 Kafka，用 upsert-kafka connector）
CREATE TABLE dim_product (
product_id STRING,
product_name STRING,
price DECIMAL(10, 2),
update_time TIMESTAMP(3),
PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'dim_product',
...
);

— Temporal Join：以订单事件时间关联商品当时的价格
SELECT o.order_id, o.amount, p.price, o.amount / p.price AS quantity
FROM ods_order o
JOIN dim_product FOR SYSTEM_TIME AS OF o.ts AS p
ON o.product_id = p.product_id;

4. DWS 层：窗口聚合

— 每分钟各城市 GMV 统计
INSERT INTO dws_city_gmv_1min
SELECT
city,
window_start,
window_end,
SUM(amount) AS gmv,
COUNT(DISTINCT user_id) AS uv
FROM TABLE(
TUMBLE(TABLE dwd_order_wide, DESCRIPTOR(ts), INTERVAL '1' MINUTES)
)
GROUP BY city, window_start, window_end;

5. 生产最佳实践

5.1 多任务资源规划

# 按 Source 并行度 = Kafka 分区数规划
source.parallelism = kafka_partitions # 通常 16-64

# 各算子并行度评估
transform.parallelism = source.parallelism # CPU 密集型，与 source 相同
window.parallelism = source.parallelism / 2 # 窗口聚合通常不是瓶颈
sink.parallelism = 4–8 # 写入端一般不需要太高

# TM 数量 = max_parallelism / slot_per_tm
# 推荐每个 TM 4-8 个 Slot

5.2 限流与降级

// Source 端限流（读 Kafka 的速率限制）
KafkaSource.<Event>builder()
.setProperty("max.poll.records", "1000") // 每批次最多拉 1000 条

// 算子级别限流（使用 RateLimiter）
RateLimiter limiter = RateLimiter.create(10000); // 10000 qps
stream.map(event -> {
limiter.acquire();
return process(event);
});

5.3 数据质量保障

// 方案1：异常数据旁路输出（sideOutput）
OutputTag<Event> badDataTag = new OutputTag<Event>("bad-data") {};

SingleOutputStreamOperator<Result> result = stream
.process(new ProcessFunction<Event, Result>() {
public void processElement(Event event, Context ctx, Collector<Result> out) {
if (!isValid(event)) {
ctx.output(badDataTag, event); // 不合法数据送到旁路
return;
}
out.collect(transform(event));
}
});

// 不合法数据写到告警 topic
result.getSideOutput(badDataTag).addSink(new KafkaSink<>(...));

// 方案2：指标监控（自定义 Metrics）
Counter invalidCount = getRuntimeContext()
.getMetricGroup()
.counter("invalid_event_count");

invalidCount.inc(); // 每次发现不合法数据就 +1
// 配合 Prometheus + Grafana 监控

5.4 重复数据处理

场景：Flink 从 CK 恢复后，Source 会从 CK 记录的 offset 重新消费，可能产生重复。

解决方案：

— 方案1：Sink 端幂等写入（ClickHouse ReplacingMergeTree）
CREATE TABLE result_table (
user_id String,
event_date Date,
cnt UInt64,
update_time DateTime
) ENGINE = ReplacingMergeTree(update_time)
PARTITION BY event_date
ORDER BY (user_id, event_date);
— ReplacingMergeTree 按主键去重（合并时保留最新）

— 方案2：Flink 内去重（ROW_NUMBER）
SELECT user_id, cnt FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts DESC) AS rn
FROM result_stream
) WHERE rn = 1;

5.5 监控告警关键指标

小结

实时数仓建设核心原则：
1. 分层：ODS → DWD → DWS → ADS，各层职责清晰
2. 维度关联：优先用 Lookup Join（无状态），业务需要历史快照时用 Temporal Join
3. 窗口优先：能用窗口就不用无界 GROUP BY
4. 状态必须有 TTL
5. Sink 幂等写入（ClickHouse ReplacingMergeTree / Upsert）
6. 完善监控：CK 状态、Lag、反压、内存