前置准备
已部署可用的 Iceberg catalog，支持
hadoop、hive 等类型，也可以通过 catalog.properties.catalog-impl 指定自定义实现。作业运行环境需能访问 warehouse 目录与元数据服务。
版本说明
Flink 版本 | 说明 |
1.20 | 基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 Iceberg 版本为 1.6、1.7、1.8、1.9、1.10 |
使用范围
可作为 Sink（
type: iceberg）使用。
快速示例
使用 Hadoop catalog：
source:type: mysqlname: MySQL Sourcehostname: 127.0.0.1port: 3306username: adminpassword: passtables: adb.\\.*server-id: 5401-5404sink:type: icebergname: Iceberg Sinkcatalog.properties.type: hadoopcatalog.properties.warehouse: /path/warehousepipeline:name: MySQL to Iceberg Pipelineparallelism: 2
使用 Hive catalog：
sink:type: icebergname: Iceberg Sinkcatalog.properties.type: hivecatalog.properties.uri: thrift://127.0.0.1:9083catalog.properties.warehouse: hdfs:///path/warehouse
Sink 配置项
参数 | 是否必填 | 默认值 | 类型 | 说明 |
catalog.properties.type | 条件必填 | (none) | String | Iceberg catalog 的类型，可取 hadoop、hive。 与 catalog.properties.catalog-impl 必须配置其一。 |
catalog.properties.catalog-impl | 条件必填 | (none) | String | 自定义 catalog 实现类的全限定名。 内置的 hadoop 与 hive 类型已覆盖常用场景，通常无需配置； 如需对接其他 catalog，可指定其实现类。 与 catalog.properties.type 必须配置其一。 |
catalog.properties.warehouse | optional | (none) | String | Iceberg 仓库的根目录，适用于所有 catalog 类型。 通常为本地或分布式文件系统路径，例如 hdfs://namenode:8020/warehouse。 |
catalog.properties.io-impl | optional | (none) | String | 自定义 FileIO 实现类的全限定名。 warehouse 不在默认文件系统上时，需要指定对应的实现类。 |
partition.key | optional | (空) | String | 为目标表指定分区键，详见「分区表」章节。 |
sink.compaction.enabled | optional | false | Boolean | 是否开启小文件合并。属于实验性能力，同步的表较多时可能造成数据流阻塞， 请谨慎开启，或改用 Iceberg 的表维护服务进行小文件合并。 |
sink.compaction.commit.interval | optional | 10 | Integer | 每张 Iceberg 表合并小文件的提交间隔。属于实验性能力， 仅在 sink.compaction.enabled 开启时生效。 |
sink.compaction.commit.parallelism | optional | -1 | Integer | 小文件合并的并行度。取 -1 时与 Sink 写入算子的并行度一致。 属于实验性能力，仅在 sink.compaction.enabled 开启时生效。 |
job.id.prefix | optional | cdc | String | 作业 ID 的前缀，用于区分不同作业。属于实验性能力，通常无需修改。 |
catalog.properties.* | optional | (none) | String | 透传给 Iceberg catalog 的参数，属性名去掉该前缀后传给 catalog。 例如 catalog.properties.uri: thrift://127.0.0.1:9083。 |
table.properties.* | optional | (none) | String | 建表时写入 Iceberg 表属性的参数，属性名去掉该前缀后作为表属性， 例如 table.properties.format-version: '2'。 |
Sink 侧的时区取自 pipeline 配置
local-time-zone，未配置时使用作业运行环境的系统时区。
分区表
通过
partition.key 可以为目标表指定分区键。多条配置之间用英文分号（;）分隔，表名与分区键之间用英文冒号（
:）分隔，同一张表的多个分区键之间用英文逗号（,）分隔。sink:type: icebergpartition.key: testdb.table1:id1,id2;testdb.table2:name
除直接指定字段名外，还支持下列分区转换：
语法 | 说明 |
identity(name)（直接写字段名） | 按字段原值分区，未匹配到任何转换语法时按此处理 |
year(name) | 按字段的年份分区 |
month(name) | 按字段的月份分区 |
day(name) | 按字段的日期分区 |
hour(name) | 按字段的小时分区 |
bucket[N](name) | 按字段哈希分到 N 个桶 |
truncate[W](name) | 按字段截断到宽度 W 后分区 |
sink:type: icebergpartition.key: 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)
需要注意：
表名需要与上游表的标识一致，格式为
namespace.tableName。格式不符合要求时作业启动失败。
配置了
partition.key 时，优先使用这里的分区键；未配置时使用上游表结构自带的分区键。
表结构变更同步
Iceberg Sink 支持下列表结构变更：
变更类型 | 说明 |
CREATE TABLE | 自动建库建表。 |
ADD COLUMN | 新增列，支持 FIRST、AFTER、BEFORE、LAST 四种位置。 |
DROP COLUMN | 删除列。 |
RENAME COLUMN | 重命名列。 |
ALTER COLUMN TYPE | 修改列类型。 |
下列表结构变更暂不支持，上游产生这些事件时作业会失败：
DROP TABLE（删除表）
TRUNCATE TABLE（清空表）
ALTER TABLE COMMENT（修改表注释）
上游存在删表、清空表、改表注释的操作时，需要在 Source 侧通过
exclude.schema.changes 排除对应事件类型，否则作业会失败。
自动建表
上游表不存在时，连接器会自动创建对应的 Iceberg 表，规则如下：
命名空间不存在时自动创建。
列的类型按下表「数据类型映射」转换，并保留列注释。
上游表的主键会被记录为 Iceberg 表的 identifier fields。
table.properties.* 指定的参数会写入表属性，对所有新建表生效。列的默认值仅在表格式为 v3 及以上时写入，v2 及以下会被忽略。
数据文件格式为 Parquet。
数据类型映射
CDC 类型 | Iceberg 类型 | 说明 |
CHAR(n) VARCHAR(n) STRING | string | - |
BOOLEAN | boolean | - |
BINARY(n) VARBINARY(n) BYTES | binary | - |
DECIMAL(p, s) | decimal(p, s) | |
TINYINT | int | 统一提升为 int。 |
SMALLINT | int | 统一提升为 int。 |
INT | int | - |
BIGINT | long | - |
FLOAT | float | - |
DOUBLE | double | - |
DATE | date | - |
TIME(p) | time | - |
TIMESTAMP(p) | timestamp | - |
TIMESTAMP_LTZ(p) TIMESTAMP WITH TIME ZONE(p) | timestamptz | - |
