3、实时写入核心 API：DataStream API 写入、Table API 写入、Flink Sink 配置、Exactly-Once 语义保证

好，咱们进入第三章。这一章是实战的硬核部分——实时写入。说白了，就是怎么把数据实时地、可靠地塞进 Iceberg 表里。

Flink 是当前最主流的实时计算引擎，Iceberg 和 Flink 的集成也是做得最成熟的。我个人习惯把写入方式分成两派：一派是 DataStream API，适合需要精细控制逻辑的场景；另一派是 Table API，适合快速开发、SQL 优先的场景。咱们一个一个来拆。

核心要点：无论你用哪种 API，最终都要落到 Flink Sink 的配置上。而配置的终极目标，就是实现 Exactly-Once 语义——数据不丢、不重、刚刚好。

3.1 DataStream API 写入：精细控制每一行

DataStream API 是 Flink 最底层的流处理接口。你想想看，如果你需要从 Kafka 消费数据，做复杂的转换，再写入 Iceberg，用 DataStream API 最灵活。

我在项目中遇到过这样一个场景：业务方要求对每条数据做字段级别的校验，不符合规则的要写到死信队列。这种逻辑用 SQL 很难表达，但用 DataStream API 就游刃有余。

先看一个最基础的写入示例：

// 1. 创建 Flink 环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 配置 checkpoint（Exactly-Once 的基础）
env.enableCheckpointing(60000); // 每分钟做一次 checkpoint
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 3. 构建 DataStream（这里假设从 Kafka 消费）
DataStream<RowData> stream = ...;

// 4. 创建 Iceberg 的 DataStream 写入器
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://path/to/warehouse/db/table");
IcebergStreamWriter<RowData> writer = FlinkSink.forRowData(stream)
    .tableLoader(tableLoader)
    .equalityFieldColumns(Arrays.asList("id")) // 指定主键，用于 upsert
    .overwrite(true) // 是否覆盖写入
    .build();

// 5. 启动任务
env.execute("Iceberg DataStream Write Job");

嗯，这里要注意几个关键点：

  • Checkpoint 必须开启：没有 checkpoint，就没有 Exactly-Once。我见过不少新手上来就写，结果数据重复了都不知道为什么。
  • equalityFieldColumns：这是做 upsert 的关键。指定哪些字段作为唯一键，Iceberg 会根据这些字段去重。
  • overwrite：如果设为 true，每次 checkpoint 会覆盖相同分区的数据。适合全量刷新场景。

3.2 Table API 写入：SQL 优先，开发效率高

如果你团队里分析师多，或者业务逻辑相对固定，Table API 是更好的选择。它让你用 SQL 就能完成写入，开发效率高出一大截。

我曾经在一个实时数仓项目里，用 Table API 把 20 多张 Kafka 主题直接映射成 Iceberg 表，整个团队只需要写 SQL，运维成本极低。

看一个典型的 Table API 写入示例：

// 1. 创建 Table 环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 2. 注册 Kafka 源表
tableEnv.executeSql("CREATE TABLE kafka_source (" +
    "  id INT, name STRING, ts TIMESTAMP(3)" +
    ") WITH (" +
    "  'connector' = 'kafka'," +
    "  'topic' = 'input_topic'," +
    "  'properties.bootstrap.servers' = 'localhost:9092'," +
    "  'format' = 'json'" +
    ")");

// 3. 注册 Iceberg 目标表
tableEnv.executeSql("CREATE TABLE iceberg_sink (" +
    "  id INT, name STRING, ts TIMESTAMP(3)" +
    ") WITH (" +
    "  'connector' = 'iceberg'," +
    "  'catalog-type' = 'hadoop'," +
    "  'warehouse' = 'hdfs://path/to/warehouse'," +
    "  'format-version' = '2'" +
    ")");

// 4. 执行写入
tableEnv.executeSql("INSERT INTO iceberg_sink SELECT * FROM kafka_source");

你看，核心就三步：建源表、建目标表、执行 INSERT INTO。Iceberg 的 Flink connector 会自动处理事务、文件提交、元数据更新。

我的建议：如果业务逻辑简单，优先用 Table API。它背后封装了 DataStream API 的复杂性，而且 Iceberg 社区对 Table API 的支持最完善。只有当你需要自定义序列化、做复杂的状态管理时，才考虑 DataStream API。

3.3 Flink Sink 配置：参数调优实战

配置 Flink Sink，说白了就是调一堆参数。我整理了一份常用配置清单，都是我在生产环境里验证过的：

参数名 默认值 推荐值 说明
write.distribution-mode hash hash 按分区键 hash 分布，避免小文件
write.target-file-size-bytes 536870912 (512MB) 134217728 (128MB) 目标文件大小，调小可以减少写入延迟
write.upsert.enabled false true 启用 upsert 模式，需要 format-version=2
commit.manifest.target-size-bytes 8388608 (8MB) 16777216 (16MB) manifest 文件大小，影响元数据性能
write.parallelism null 并行度 写入并行度，建议与 source 并行度一致

这里我重点说几个坑：

  • write.target-file-size-bytes 别设太大：我曾经设成 1GB，结果 checkpoint 间隔内文件写不满，导致频繁提交小文件，元数据膨胀得厉害。
  • write.upsert.enabled 必须配合 format-version=2：V1 格式不支持行级删除，upsert 会报错。这个坑我踩过，排查了半天才发现是格式版本不对。
  • write.distribution-mode 默认是 hash：如果你的表没有分区键，hash 分布会导致所有数据写到同一个文件，反而成了瓶颈。这时候可以改成 none。

3.4 Exactly-Once 语义保证：原理与实战

Exactly-Once 是实时写入的圣杯。Iceberg 是怎么做到的？说白了，靠两样东西：Flink 的 checkpoint 机制 + Iceberg 的事务性提交。

流程是这样的：

  1. Flink 定期做 checkpoint，把当前状态和写入的文件列表保存下来。
  2. Iceberg 的 writer 在 checkpoint 完成时，会提交一批新文件到 Iceberg 表。
  3. 如果任务失败，Flink 从最近的 checkpoint 恢复，重新写入那些未提交的文件。
  4. Iceberg 通过元数据中的文件列表，保证同一批数据不会重复提交。

我画了一张流程图，帮你理解这个过程：

Iceberg Exactly-Once 写入流程 Flink Source Iceberg Writer Checkpoint 任务失败 从 Checkpoint 恢复 重新写入未提交数据 Iceberg 事务提交 关键：Iceberg 的事务提交是原子的。要么全部提交成功，要么全部回滚。这保证了数据不会出现部分写入的情况。

为什么会这样设计？因为 Iceberg 的元数据层是独立的。文件写入和元数据更新是两回事。Flink 负责把文件写出来，Iceberg 负责决定哪些文件对查询可见。这两者解耦了，Exactly-Once 才有可能实现。

注意：Exactly-Once 不是免费的。它需要额外的存储和计算开销。具体来说：

  • Checkpoint 状态会占用状态后端存储（RocksDB 或 Heap）
  • Iceberg 的元数据文件会随着 checkpoint 次数增加而增长
  • 任务恢复时，需要重新处理从上次 checkpoint 到失败之间的数据

所以，如果你的业务允许少量数据重复（比如一些统计报表），可以考虑用 At-Least-Once 语义，性能会好很多。

3.5 三种写入模式对比

最后，我把三种常见的写入模式做个对比，方便你选型：

模式 适用场景 优点 缺点
Append 日志、事件流、不可变数据 写入快，元数据开销小 不支持更新和删除
Upsert 维表、CDC 数据、需要去重的场景 支持行级更新，数据一致性好 需要 format-version=2，写入性能略低
Overwrite 全量刷新、分区替换 简单粗暴，适合批处理 会覆盖整个分区，不适合增量场景

嗯，这一章的内容就到这里。记住一句话：实时写入的核心不是写，而是怎么保证写对了。DataStream API 给你控制力，Table API 给你效率，Flink Sink 配置给你调优空间，Exactly-Once 给你信心。把这四样东西吃透了，Iceberg 实时写入这块你就拿下了。

← 上一章 📖 返回目录 下一章 →