Iceberg Pipeline 连接器

Iceberg Pipeline 连接器作为数据管道的 Data Sink，支持将数据写入 Apache Iceberg 表。本文档介绍如何配置该连接器。

核心能力

• 自动建表： 当 Iceberg 表不存在时，自动动态创建
• Schema 同步： 将上游数据源的 Schema 变更（例如新增列）自动同步到 Iceberg 表
• 数据同步： 支持批处理和流式数据同步

Pipeline 创建方式

以下示例展示了如何定义一个从 MySQL 读取数据并写入 Iceberg 的数据管道：

Hadoop Catalog 示例

source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
  server-id: 5401-5404

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.type: hadoop
  catalog.properties.warehouse: /path/warehouse

pipeline:
  name: MySQL to Iceberg Pipeline
  parallelism: 2

AWS Glue Catalog 示例

source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
  server-id: 5401-5404

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.type: glue
  catalog.properties.warehouse: s3://my-bucket/warehouse
  catalog.properties.io-impl: org.apache.iceberg.aws.s3.S3FileIO
  catalog.properties.client.region: us-east-1
  catalog.properties.glue.skip-archive: true

pipeline:
  name: MySQL to Iceberg via Glue Pipeline
  parallelism: 2

注意： 根据所使用的 Catalog 类型，可能需要手动添加额外的 JAR 依赖，并在使用 Flink CDC CLI 提交 YAML 管道作业时通过 --jar 参数传入。

Catalog 类型	依赖项	说明
所有类型	org.apache.iceberg:iceberg-flink-runtime-1.20	Iceberg Flink 运行时依赖。当运行环境中未预装 Iceberg 时（例如独立部署的 Flink 集群），所有 Catalog 类型均需要此依赖。
hadoop	org.apache.flink:flink-shaded-hadoop-2-uber:2.8.3-10.0	提供 Hadoop 文件系统相关依赖。
glue	org.apache.iceberg:iceberg-aws	提供 AWS Glue Catalog 和 S3 FileIO 实现。
glue	software.amazon.awssdk:bundle	iceberg-aws 所依赖的 AWS SDK。

Pipeline 连接器选项

选项	是否必填	默认值	类型	描述
type	必填	（无）	String	指定使用的连接器类型，此处应为 iceberg。
name	可选	（无）	String	Sink 的名称。
catalog.properties.type	条件必填	（无）	String	Iceberg Catalog 的元数据存储类型，支持 hadoop、hive 和 glue。此选项与 catalog.properties.catalog-impl 必须设置其一。
catalog.properties.catalog-impl	条件必填	（无）	String	自定义 Catalog 实现类，例如 org.apache.iceberg.aws.glue.GlueCatalog。此选项与 catalog.properties.type 必须设置其一。
catalog.properties.warehouse	可选	（无）	String	Iceberg Catalog 的仓库根路径，适用于所有 Catalog 类型。对于 hadoop 和 hive Catalog，通常为本地或分布式文件系统路径；对于 glue Catalog，通常为对象存储路径，例如 s3://my-bucket/warehouse。
catalog.properties.uri	可选	（无）	String	元数据服务 URI（例如 Hive Metastore 的 thrift URI）。
catalog.properties.io-impl	可选	（无）	String	自定义 FileIO 实现类。使用 AWS S3 时，请设置为 org.apache.iceberg.aws.s3.S3FileIO。
catalog.properties.client.region	可选	（无）	String	Glue Catalog 客户端的 AWS 区域（例如 us-east-1）。
catalog.properties.glue.id	可选	（无）	String	Glue Catalog ID（即 AWS 账户 ID）。默认使用调用者的 AWS 账户 ID。
catalog.properties.glue.skip-archive	可选	true	Boolean	是否跳过在 Glue 中归档旧版本的表元数据。
catalog.properties.glue.skip-name-validation	可选	false	Boolean	是否跳过 Glue Catalog 的名称校验。
partition.key	可选	（无）	String	各分区表的分区键。支持为多张表设置不同的分区键，表之间以 ; 分隔，分区键之间以 , 分隔。例如，可以通过 testdb.table1:id1,id2;testdb.table2:name 为两张表设置分区键。对于分区转换，可以使用以下语法：testdb.table1:truncate[10](id);testdb.table2:hour(create_time);testdb.table3:day(create_time);testdb.table4:month(create_time);testdb.table5:year(create_time);testdb.table6:bucket[10](create_time)。
catalog.properties.*	可选	（无）	String	透传 Iceberg Catalog 选项到管道，详见 Iceberg Catalog 配置。
table.properties.*	可选	（无）	String	透传 Iceberg 表选项到管道，详见 Iceberg 表配置。
hadoop.conf.*	optional	(none)	String	传递 Hadoop Configuration 参数（用于 Iceberg 的 catalog/table 相关操作）。前缀 hadoop.conf. 会被剥离。例如 hadoop.conf.fs.s3a.endpoint。

使用须知

• 源表必须包含主键，不支持无主键的表。

• 不支持精确一次（Exactly-Once）语义。连接器采用至少一次（At-Least-Once）+ 主键幂等写入的方式保证数据一致性。

数据类型映射

CDC 类型	Iceberg 类型	说明
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)
BOOLEAN	BOOLEAN
DATE	DATE
TIMESTAMP	TIMESTAMP
TIMESTAMP_LTZ	TIMESTAMP_LTZ
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)