功能概述 #

Iceberg CDC Pipeline Sink Connector 用于将 Flink CDC Pipeline 产生的数据和表结构变更写入 Apache Iceberg 表。Connector 提供普通 Iceberg 和 Iceberg on LAS 两种写入方式，支持目标表不存在时自动建表，并可持续同步具有主键的源表数据。

版本兼容性 #

特色功能 #

• 自动建表：目标 Iceberg 表不存在时，根据上游表结构创建目标表。
• Schema 同步：将 Connector 支持的上游 Schema Change 应用到目标 Iceberg 表。
• 批流一体：支持批式和流式 Pipeline 数据同步。
• 双写入路径：同一套 Pipeline 配置模式覆盖普通 Iceberg 和 Iceberg on LAS。
• Catalog 与 Table 参数透传：可通过 catalog.properties.* 和 table.properties.* 配置 Iceberg 行为。
• Upsert 自动建表：目标表不存在时创建 Iceberg Format V2 主键表并使用 Upsert 写入。

前提条件 #

1. 已开通火山引擎流式计算 Flink 版，并创建可用的 Flink 资源池。
2. Source 表包含非空主键；无主键表不能写入。
3. Flink 资源池能够访问 Iceberg Catalog、Metastore 和底层对象存储。
4. 已准备具有 Catalog 元数据和对象存储读写权限的身份凭证。
5. 使用 Hive Catalog 时，已准备 Hive Metastore 地址和 Warehouse 路径。
6. 使用 Iceberg on LAS 时，已准备 LAS Catalog、地域、Metastore 地址及格式正确的 hive-site.xml，并将该文件作为作业依赖分发。
7. 已将 LAS AK/SK 直接填写到 hive-site.xml。该文件不支持产品变量，需限制文件访问权限并避免将其提交到代码仓库。

使用限制 #

• 只支持具有主键的 Source 表。
• 只支持 Iceberg Format V2 Upsert 表，不支持 Append 表。
• Connector 不提供 Exactly-Once 交付语义，而是采用 At-Least-Once，并通过 Iceberg 主键表实现幂等写入。
• Schema Evolution 可处理建表、新增列、修改列类型、重命名列、删除列等事件。注意当前版本暂不支持同步清空表和删除表。
• partition.key 仅配置直接分区字段，当前暂不支持 truncate、hour、day、month、year、bucket 等 Partition Transform 写法。
• Iceberg on LAS 依赖 hive-site.xml。XML 格式错误、文件未随作业分发或配置路径错误都会导致 Catalog 初始化失败。
• 普通 Iceberg 与 Iceberg on LAS 使用不同的 type 值，不能混用。

语法结构 #

普通 Iceberg #

以下结构使用 Hive Catalog：

source:
  type: <source_type>
  name: <source_name> 
  # Source 参数

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.type: hive
  catalog.properties.uri: thrift://<hive_metastore_host>:<port>
  catalog.properties.warehouse: tos://<bucket>/<warehouse_path>

pipeline:
  name: <pipeline_name>
  parallelism: 2

Iceberg on LAS #

source:
  type: <source_type> 
  name: <source_name>
  # Source 参数

sink:
  type: iceberg-las
  name: Iceberg LAS Sink
  catalog.properties.type: hive
  catalog.properties.uri: ${iceberg_catalog.properties.uri}
  catalog.properties.warehouse: ${iceberg_catalog.properties.warehouse}
  catalog.properties.hive-conf-dir: /opt/tiger/workdir

pipeline:
  name: <pipeline_name>
  parallelism: 2

参数说明 #

通用参数 #

Catalog 与 Table 参数 #

Iceberg on LAS 配置 #

以下配置写入随作业分发的 hive-site.xml：

类型映射 #

使用示例 #

将 MySQL CDC 数据写入普通 Iceberg #

source:
  type: mysql
  name: MySQL Source
  hostname: <mysql_host>
  port: 3306
  username: ${secret_values.mysql-username}
  password: ${secret_values.mysql-password}
  tables: <database_name>.<table_name>
  server-id: 5401-5404

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.type: hive
  catalog.properties.uri: thrift://<hive_metastore_host>:9083
  catalog.properties.warehouse: tos://<bucket>/<warehouse_path>

pipeline:
  name: MySQL to Iceberg Pipeline
  parallelism: 2

预期结果：Pipeline 在目标 Catalog 中创建与源表对应的主键表，写入初始快照，并持续应用新增、更新和删除事件。

将 MySQL CDC 数据写入 Iceberg on LAS #

首先准备 hive-site.xml：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
  <property>
    <name>metastore.catalog.default</name>
    <value><las_catalog_name></value>
  </property>
  <property>
    <name>hive.client.las.region.name</name>
    <value><region></value>
  </property>
  <property>
    <name>hive.metastore.uris</name>
    <value>thrift://<las_metastore_host>:<port></value>
  </property>
  <property>
    <name>hive.hms.client.is.public.cloud</name>
    <value>true</value>
  </property>
  <property>
    <name>hive.client.las.ak</name>
    <value><access_key></value>
  </property>
  <property>
    <name>hive.client.las.sk</name>
    <value><secret_key></value>
  </property>
</configuration>

将 <access_key>、<secret_key> 替换为真实 LAS AK/SK，再把 hive-site.xml 作为作业依赖上传。该文件不支持变量；catalog.properties.hive-conf-dir 固定为 /opt/tiger/workdir：

source:
  type: mysql
  name: MySQL Source
  hostname: <mysql_host>
  port: 3306
  username: ${secret_values.mysql-username}
  password: ${secret_values.mysql-password}
  tables: <database_name>.<table_name>
  server-id: 5401-5404

sink:
  type: iceberg-las
  name: Iceberg LAS Sink
  catalog.properties.type: hive
  catalog.properties.uri: ${iceberg_catalog.properties.uri}
  catalog.properties.warehouse: ${iceberg_catalog.properties.warehouse}
  catalog.properties.hive-conf-dir: /opt/tiger/workdir

pipeline:
  name: MySQL to Iceberg LAS Pipeline
  parallelism: 2

预期结果：Pipeline 通过 LAS Metastore 创建或访问目标 Iceberg 表，并将数据文件写入配置的对象存储 Warehouse。

运行机制 #

Connector 接收 Source 产生的数据变更和 Schema Change。目标表不存在时，Connector 根据上游表结构创建 Iceberg Format V2 主键表并自动使用 Upsert；目标表存在时，将数据变更写入对应的 Upsert 表。Append 表不受支持。
数据写入采用 At-Least-Once。作业恢复时，同一变更可能被重新提交，Connector 依赖表主键覆盖相同记录以实现幂等。因此，Source 主键必须稳定，且不能在同步过程中产生重复或为空的业务标识。
Schema Evolution 行为与 Flink CDC 3.4 保持一致。可以通过 schema.change.behavior 控制异常、应用、尝试应用、宽松或忽略模式，并通过 include.schema.changes、exclude.schema.changes 过滤具体事件。涉及删除列、清空表或删除表的配置可能造成数据丢失，上线前应进行专项验证。

验证与排障 #

验证方法 #

1. 提交作业并确认运行状态正常，日志中没有 Catalog、Metastore、对象存储或权限错误。
2. 确认目标 Catalog 中已创建对应 Iceberg 主键表，字段和主键与上游一致。
3. 核对初始快照完成后的目标表记录数和关键字段。
4. 对同一主键执行新增、更新和删除，确认目标表结果与源端一致。
5. 触发一个已支持的 Schema Change，确认目标表 Schema 正确更新。
6. 等待成功 Checkpoint 后执行受控重启，确认恢复后最终数据正确且没有重复主键。
7. 使用 Iceberg on LAS 时，同时检查 LAS Catalog 元数据和对象存储数据文件。

常见问题 #

注意事项 #

• 不要在 Pipeline YAML、公开文档、日志或代码仓库中保存真实 AK/SK、数据库密码或对象存储凭证。
• Iceberg on LAS 必须在 hive-site.xml 中直接填写 AK/SK，不支持产品变量。应限制该文件的访问与分发范围，并按组织安全要求定期轮换凭证。
• catalog.properties.hive-conf-dir 固定为 /opt/tiger/workdir，不要改为本地上传路径或其他运行目录。
• Warehouse 路径设置错误可能将数据写入非预期位置；上线前应在 LAS 或对象存储控制台核对实际路径。
• Schema Evolution 上线前应在测试表验证，避免不兼容变更影响整个 Pipeline。

相关文档 #

• Apache Flink CDC 3.4：Iceberg Pipeline Connector
• Apache Flink CDC 3.4：Schema Evolution
• Apache Iceberg：Flink Writes
• 火山引擎流式计算 Flink 版：开发 Flink CDC 任务
• 火山引擎流式计算 Flink 版：使用变量