千亿级流量下的实时湖仓一体架构：Flink + Iceberg + Paimon 全链路实战与性能调优

MySQL (Binlog) → Canal 1.1.6 → Kafka 3.4.0 (partition=32, retention=7d)
                    ↓
              Flink 1.17.1 (Checkpoint 间隔 30s, Exactly-Once)
                    ↓
        ┌───────────┴───────────┐
        │                       │
   Iceberg 1.4.0           Paimon 0.7.0
   (ODS/DWD 层)           (DWS 层，主键聚合表)
        │                       │
        └───────────┬───────────┘
                    ↓
           Trino 422 (或 StarRocks 3.2)
                    ↓
             BI 看板 / API 服务

<flink.version>1.17.1</flink.version>
<iceberg.version>1.4.0</iceberg.version>
<paimon.version>0.7.0-incubating</paimon.version>

<!-- Flink SQL 连接器 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-sql-connector-kafka</artifactId>
    <version>3.0.2-1.17</version>
</dependency>
<dependency>
    <groupId>org.apache.iceberg</groupId>
    <artifactId>iceberg-flink-runtime-1.17</artifactId>
    <version>${iceberg.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.paimon</groupId>
    <artifactId>paimon-flink-1.17</artifactId>
    <version>${paimon.version}</version>
</dependency>

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(30_000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:8020/flink/checkpoints");
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10_000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);    // 容忍 3 次失败
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

CREATE TABLE order_cdc (
    order_id        BIGINT,
    user_id         BIGINT,
    product_id      BIGINT,
    amount          DECIMAL(12,2),
    order_time      TIMESTAMP(3) METADATA FROM 'timestamp' VIRTUAL,
    database_name   STRING METADATA FROM 'database_name' VIRTUAL,
    table_name      STRING METADATA FROM 'table_name' VIRTUAL,
    op_ts           TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'order_topic',
    'properties.bootstrap.servers' = 'kafka-1:9092,kafka-2:9092',
    'properties.group.id' = 'flink-iceberg-ods',
    'scan.startup.mode' = 'group-offsets',          -- 从 Kafka 消费组 offset 开始
    'format' = 'debezium-json',
    'debezium-json.schema-include' = 'true',
    'debezium-json.ignore-parse-errors' = 'true',   -- 容忍个别脏数据
    'properties.fetch.max.bytes' = '52428800'       -- 50MB，避免大消息超时
);

CREATE CATALOG iceberg_catalog WITH (
    'type' = 'iceberg',
    'catalog-type' = 'hive',
    'uri' = 'thrift://hms:9083',
    'warehouse' = 's3a://data-lake/warehouse',
    'cache-enabled' = 'true',
    'cache.expiration-interval-ms' = '30000'
);

CREATE TABLE iceberg_catalog.ods.order_ods (
    order_id        BIGINT,
    user_id         BIGINT,
    product_id      BIGINT,
    amount          DECIMAL(12,2),
    event_time      TIMESTAMP(3),
    op_type         STRING,          -- 'INSERT' / 'UPDATE' / 'DELETE'
    _source_ts      TIMESTAMP_LTZ(3) -- 摄入时间
) PARTITIONED BY (days(event_time))   -- 按天分区
WITH (
    'format-version' = '2',
    'write.upsert.enabled' = 'false',             -- 纯 Append，不做行级更新
    'write.distribution-mode' = 'hash',
    'write.target-file-size-bytes' = '268435456', -- 256MB
    'write.parquet.compression-codec' = 'zstd',
    'write.parquet.compression-level' = '3',
    'write.metadata.delete-after-commit.enabled' = 'true',
    'write.metadata.previous-versions-max' = '10' -- 保留最近 10 个 metadata 版本
);

INSERT INTO iceberg_catalog.ods.order_ods
SELECT 
    order_id,
    user_id,
    product_id,
    amount,
    order_time AS event_time,
    CASE 
        WHEN op_ts IS NOT NULL AND op_ts > CURRENT_TIMESTAMP - INTERVAL '5' SECOND 
        THEN 'INSERT' 
        ELSE 'UPDATE' 
    END AS op_type,
    CURRENT_TIMESTAMP AS _source_ts
FROM order_cdc;

CALL iceberg_catalog.system.rewrite_data_files(
    table => 'ods.order_ods',
    options => map(
        'strategy', 'sort',
        'sort-order', 'event_time ASC',
        'max-file-group-size-bytes', '536870912'   -- 512MB
    )
);
CALL iceberg_catalog.system.expire_snapshots(
    table => 'ods.order_ods',
    older_than => TIMESTAMP '2026-08-01 00:00:00',
    retain_last => 5
);

CREATE CATALOG paimon_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs://namenode:8020/paimon/warehouse'
);

CREATE TABLE paimon_catalog.dws.product_uv_daily (
    product_id   BIGINT,
    stat_date    STRING,          -- yyyyMMdd
    uv_bitmap    BYTES,           -- 存储 RoaringBitmap 序列化
    uv_count     BIGINT,
    PRIMARY KEY (product_id, stat_date) NOT ENFORCED
) WITH (
    'bucket' = '16',                         -- 分桶数，按 product_id 打散
    'changelog-producer' = 'full-compaction', -- 仅在 compaction 时产生 changelog
    'merge-engine' = 'aggregation',
    'fields.uv_bitmap.aggregate-function' = 'bitmap_union',
    'fields.uv_count.aggregate-function' = 'sum',
    'compaction.max-file-num' = '30',
    'compaction.target-file-size' = '256MB'
);

import org.apache.flink.table.functions.AggregateFunction;
import org.roaringbitmap.RoaringBitmap;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;

public class BitmapAggFunction extends AggregateFunction<byte[], RoaringBitmap> {

    @Override
    public RoaringBitmap createAccumulator() {
        return new RoaringBitmap();
    }

    @Override
    public byte[] getValue(RoaringBitmap acc) {
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
             DataOutputStream dos = new DataOutputStream(baos)) {
            acc.serialize(dos);
            dos.flush();
            return baos.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException("Failed to serialize RoaringBitmap", e);
        }
    }

    public void accumulate(RoaringBitmap acc, Long userId) {
        if (userId != null && userId > 0 && userId <= Integer.MAX_VALUE) {
            acc.add(userId.intValue());
        }
    }

    public void merge(RoaringBitmap acc, Iterable<RoaringBitmap> iterable) {
        for (RoaringBitmap other : iterable) {
            acc.or(other);
        }
    }

    public void resetAccumulator(RoaringBitmap acc) {
        acc.clear();
    }
}

CREATE FUNCTION bitmap_agg AS 'com.xxx.BitmapAggFunction';

INSERT INTO paimon_catalog.dws.product_uv_daily
SELECT 
    product_id,
    DATE_FORMAT(event_time, 'yyyyMMdd') AS stat_date,
    bitmap_agg(user_id) AS uv_bitmap,
    CAST(0 AS BIGINT) AS uv_count   -- 占位，由 Paimon 自动 sum 聚合
FROM iceberg_catalog.ods.order_ods
WHERE op_type = 'INSERT'  -- 只统计新增订单
GROUP BY product_id, DATE_FORMAT(event_time, 'yyyyMMdd');

-- 获取 ODS 表的最新快照 ID
SET snapshot_id = (SELECT snapshot_id FROM iceberg_catalog.ods.order_ods.snapshots ORDER BY committed_at DESC LIMIT 1);

-- 插入宽表（仅增量部分）
INSERT INTO iceberg_catalog.dwd.order_detail_wide
SELECT 
    o.order_id,
    u.user_name,
    p.product_name,
    p.category,
    o.amount,
    o.event_time
FROM (
    SELECT * FROM iceberg_catalog.ods.order_ods
    /*+ OPTIONS('snapshot-id' = ${snapshot_id}) */
) o
LEFT JOIN iceberg_catalog.dim.user_dim 
    /*+ OPTIONS('snapshot-id' = (SELECT snapshot_id FROM iceberg_catalog.dim.user_dim.snapshots ORDER BY committed_at DESC LIMIT 1)) */
    FOR SYSTEM_TIME AS OF o.event_time
ON o.user_id = u.user_id
LEFT JOIN iceberg_catalog.dim.product_dim 
    FOR SYSTEM_TIME AS OF o.event_time
ON o.product_id = p.product_id
WHERE o.event_time > CURRENT_TIMESTAMP - INTERVAL '10' MINUTE;   -- 只取最近 10 分钟数据

'compaction.strategy' = 'level',
'compaction.level0-file-num' = '10',
'compaction.level1-file-num' = '20',
'compaction.target-file-size' = '256MB',
'compaction.max-file-num' = '50'

-- 第一阶段：加盐打散
CREATE VIEW pre_agg_salt AS
SELECT 
    product_id,
    MOD(product_id, 10) AS salt,
    DATE_FORMAT(event_time, 'yyyyMMdd') AS stat_date,
    bitmap_agg(user_id) AS uv_bitmap_partial
FROM ods_order
GROUP BY product_id, MOD(product_id, 10), DATE_FORMAT(event_time, 'yyyyMMdd');

-- 第二阶段：去盐合并
INSERT INTO dws_product_uv
SELECT 
    product_id,
    stat_date,
    bitmap_union(uv_bitmap_partial) AS uv_bitmap,
    CAST(0 AS BIGINT) AS uv_count
FROM pre_agg_salt
GROUP BY product_id, stat_date;

-- 从 MySQL 直接查询订单总数
SELECT COUNT(*) FROM order_source WHERE dt = '${yesterday}';

-- 从 Iceberg ODS 查询
SELECT COUNT(*) FROM iceberg_catalog.ods.order_ods WHERE days(event_time) = '${yesterday}';

-- 从 Paimon DWS 聚合 UV 与 MySQL 去重后的 UV 对比