大数据架构师实战：从 Lambda 到湖仓一体的架构演进与代码实现

-- 创建 Kafka 源表（ODS 层）
CREATE TABLE ods_user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'raw_events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'flink_ods_consumer',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);

-- 创建 DWD 层明细表（写入 Paimon）
CREATE TABLE dwd_user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    event_time TIMESTAMP(3),
    user_region STRING,
    item_category STRING,
    PRIMARY KEY (user_id, event_time) NOT ENFORCED
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://warehouse/paimon/dwd_user_behavior',
    'bucket' = '8',
    'merge-engine' = 'aggregation'
);

-- 实时 ETL：清洗 + 维表关联
INSERT INTO dwd_user_behavior
SELECT 
    o.user_id,
    o.item_id,
    o.behavior,
    o.event_time,
    u.region AS user_region,
    i.category AS item_category
FROM ods_user_behavior o
LEFT JOIN dim_user u ON o.user_id = u.user_id
LEFT JOIN dim_item i ON o.item_id = i.item_id
WHERE o.behavior IS NOT NULL;

-- 创建 DWS 层聚合表
CREATE TABLE dws_hourly_agg (
    hour_start TIMESTAMP(3),
    item_id BIGINT,
    pv BIGINT,
    uv BIGINT,
    PRIMARY KEY (hour_start, item_id) NOT ENFORCED
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://warehouse/paimon/dws_hourly_agg',
    'bucket' = '16',
    'merge-engine' = 'aggregation'
);

-- 小时级聚合（Flink 流式写入）
INSERT INTO dws_hourly_agg
SELECT 
    DATE_FORMAT(event_time, 'yyyy-MM-dd HH:00:00') AS hour_start,
    item_id,
    COUNT(*) AS pv,
    COUNT(DISTINCT user_id) AS uv
FROM dwd_user_behavior
GROUP BY DATE_FORMAT(event_time, 'yyyy-MM-dd HH:00:00'), item_id;

# Flink Checkpoint 配置（保障 Exactly-Once）
flink.state.backend: rocksdb
flink.checkpoints.interval: 60s
flink.checkpoints.dir: hdfs://checkpoint_path
flink.state.backend.rocksdb.localdir: /data/rocksdb