4、实时查询与 CDC：Flink 实时读取 Iceberg、Streaming Read 原理、CDC 入湖方案、Changelog 视图

好，咱们进入第四章。这一章聊的是实时查询和 CDC，说白了就是怎么让数据「活」起来。

前面几章我们讲了 Iceberg 的批读写、表结构设计，那都是离线场景。但很多业务场景下，数据是源源不断流进来的——比如数据库的 binlog、点击流日志、IoT 设备上报。这时候就需要 Flink + Iceberg 这套组合拳了。

我个人习惯把这一章分成四个部分：Flink 怎么实时读 Iceberg、Streaming Read 的底层原理、CDC 入湖的几种常见方案、以及 Changelog 视图到底解决了什么问题。咱们一个一个来。

4.1 Flink 实时读取 Iceberg

先看最基础的操作：用 Flink SQL 实时读 Iceberg 表。

你可能会问：「Iceberg 不是存文件的吗？文件系统怎么做到实时读？」

嗯，这里的关键在于 Iceberg 的 快照（Snapshot）机制。每次写入都会生成一个新的快照，Flink 可以监听这些快照的变化，增量拉取新数据。

看一段最简单的 Flink SQL：

-- 创建一个 Iceberg 表，指定为流式读取
CREATE TABLE orders_iceberg (
    order_id BIGINT,
    user_id BIGINT,
    amount DOUBLE,
    order_ts TIMESTAMP(3)
) WITH (
    'connector' = 'iceberg',
    'catalog-database' = 'default',
    'catalog-table' = 'orders',
    'streaming' = 'true',
    'monitor-interval' = '1s'
);

-- 实时查询
SELECT * FROM orders_iceberg
/*+ OPTIONS('streaming'='true', 'starting-strategy'='earliest') */;

这段代码里，monitor-interval 控制轮询快照的频率。我建议生产环境不要设得太短，1 秒已经够快了，再短反而会给文件系统造成压力。

4.2 Streaming Read 原理

表面上看，Flink 只是每隔一秒查一下有没有新快照。但底层其实做了不少优化。

我画了一张图，帮你理解 Streaming Read 的完整链路：

流程其实不复杂：

1. 数据写入 Iceberg 表，生成新快照 S1、S2、S3……
2. Flink 定时轮询 Iceberg 的元数据，发现新快照。
3. 只读取新快照中新增或修改的文件，不碰历史数据。
4. 将增量数据发送到下游算子。

这里有个细节：Iceberg 的 Streaming Read 并不是真正的「流」，而是「微批」。每次轮询都是一次小批量的文件扫描。但好处是——它保证了 Exactly-Once 语义，不会丢数据，也不会重复。

4.3 CDC 入湖方案

CDC（Change Data Capture）是实时入湖最常见的场景。说白了，就是把数据库的 binlog 实时同步到 Iceberg 表里。

我在项目中遇到过好几次这样的需求：MySQL 的业务表每天几百万条变更，需要实时同步到数据湖做分析。最开始用 Debezium + Kafka，后来发现 Iceberg 直接支持 Flink CDC connector，省掉了 Kafka 这一层。

常见的 CDC 入湖方案有几种：

我个人最推荐第一种方案——Flink CDC + Iceberg。原因很简单：组件少，运维成本低，而且 Flink CDC 2.0 之后支持了动态加表，不用重启作业。

看一个完整的 CDC 入湖示例：

-- 1. 定义 MySQL CDC 源表
CREATE TABLE mysql_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DOUBLE,
    order_ts TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = 'password',
    'database-name' = 'shop',
    'table-name' = 'orders',
    'scan.startup.mode' = 'latest-offset'
);

-- 2. 定义 Iceberg 目标表
CREATE TABLE iceberg_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DOUBLE,
    order_ts TIMESTAMP(3),
    _event_type STRING
) WITH (
    'connector' = 'iceberg',
    'catalog-database' = 'default',
    'catalog-table' = 'orders_cdc',
    'format-version' = '2',
    'write.upsert.enabled' = 'true'
);

-- 3. 启动同步作业
INSERT INTO iceberg_orders
SELECT order_id, user_id, amount, order_ts, 'INSERT' FROM mysql_orders;

这段代码里，write.upsert.enabled=true 是关键。它告诉 Iceberg：这张表要做 Upsert，也就是根据主键去重。CDC 数据里同一条记录可能被更新多次，不开 Upsert 的话，Iceberg 里会保留多条重复记录。

4.4 Changelog 视图

好，最后一个话题——Changelog 视图。这个概念很多人第一次接触会觉得绕，我尽量讲清楚。

你想想看，CDC 数据流里包含三种操作：INSERT、UPDATE、DELETE。但 Iceberg 本身是只追加的，它没有「删除」和「更新」的概念。那怎么把 CDC 的语义映射到 Iceberg 上呢？

Changelog 视图就是干这个的。它把 Iceberg 表的数据「伪装」成一个 changelog 流，每条记录都带一个 row_kind 标记，告诉下游这条数据是新增、更新还是删除。

看个例子：

-- 创建 Changelog 视图
CREATE VIEW orders_changelog AS
SELECT order_id, user_id, amount, order_ts, row_kind
FROM iceberg_orders
/*+ OPTIONS('streaming'='true', 'changelog'='true') */;

-- 查询 changelog
SELECT * FROM orders_changelog;

输出结果大概长这样：

这里 +I 表示 INSERT，-U 表示更新前的旧值，+U 表示更新后的新值，-D 表示 DELETE。

我曾经踩过一个坑：Changelog 视图默认只保留最近一次快照的变更。如果你需要回溯历史变更，得自己维护一个 changelog 表，或者用 Iceberg 的 snapshot-id 手动指定快照。

最后总结一下这一章的核心：

• Streaming Read 本质是微批轮询，靠快照机制实现增量读取。
• CDC 入湖 推荐 Flink CDC + Iceberg 直连方案，记得开 Upsert。
• Changelog 视图 把 Iceberg 的追加语义映射成 CDC 语义，方便下游消费。