一、前置准备
·已开通腾讯云 DLC 服务，并创建好数据目录（catalog）与数据库。
·已准备腾讯云 API 密钥（SecretId / SecretKey），具备 DLC 与 COS 的访问权限。
·已获取账号 APPID 与 DLC 所在地域。
二、版本说明
| Flink 版本 | 说明 |
| 1.20 | 基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持腾讯云 DLC 托管的 Iceberg 表 |
三、使用范围
可作为 Sink（type: iceberg-dlc）使用。
四、快速示例
从 MySQL 同步到 DLC：
source: type: mysql name: MySQL Source hostname: 127.0.0.1 port: 3306 username: admin password: pass tables: adb.\\.* server-id: 5401-5404sink: type: iceberg-dlc name: DLC Sink catalog.properties.uri: dlc.internal.tencentcloudapi.com catalog.properties.qcloud.dlc.region: ap-guangzhou catalog.properties.qcloud.dlc.secret-id: your-secret-id catalog.properties.qcloud.dlc.secret-key: your-secret-key catalog.properties.qcloud.dlc.user.appid: your-appidpipeline: name: MySQL to DLC Pipeline parallelism: 2
不配置 catalog.properties.warehouse 时，使用 DLC 托管表，表数据的存储位置由 DLC 服务端分配。
如果需要自行指定存储位置，可以配置 catalog.properties.warehouse，
取值必须使用 cosn:// 或 lakefs:// 协议：
sink: type: iceberg-dlc name: DLC Sink catalog.properties.qcloud.dlc.region: ap-guangzhou catalog.properties.qcloud.dlc.secret-id: your-secret-id catalog.properties.qcloud.dlc.secret-key: your-secret-key catalog.properties.qcloud.dlc.user.appid: your-appid catalog.properties.warehouse: lakefs://your-bucket/warehouse
五、Sink 配置项
| 参数 | 是否必填 | 默认值 | 类型 | 说明 |
| catalog.properties.qcloud.dlc.region | required | (none) | String | DLC 所在地域，例如 ap-guangzhou。 |
| catalog.properties.qcloud.dlc.secret-id | required | (none) | String | 腾讯云 API 密钥的 SecretId。 |
| catalog.properties.qcloud.dlc.secret-key | required | (none) | String | 腾讯云 API 密钥的 SecretKey。 |
| catalog.properties.qcloud.dlc.user.appid | required | (none) | String | 腾讯云账号 APPID。 |
| catalog.properties.uri | optional | dlc.internal.tencentcloudapi.com | String | DLC 服务端点地址。 |
| catalog.properties.qcloud.dlc.managed.account.uid | optional | 100026378089 | String | DLC 托管账号 UID。 |
| catalog.properties.qcloud.dlc.credential.provider.class | optional | (none) | String | 自定义凭证提供类的全限定名。不配置时使用上述密钥参数构造凭证。 |
| catalog.properties.warehouse | optional | (none) | String | 仓库根目录，必须使用 cosn:// 或 lakefs:// 协议， 其它协议会导致作业启动失败。未配置时使用 DLC 托管表，存储位置由 DLC 服务端分配。 |
| catalog.properties.fs.lakefs.impl | optional | org.apache.hadoop.fs.CosFileSystem | String | lakefs 协议使用的 Hadoop 文件系统实现类。 |
| catalog.properties.fs.cosn.impl | optional | org.apache.hadoop.fs.CosFileSystem | String | cosn 协议使用的 Hadoop 文件系统实现类。 |
| catalog.properties.fs.cosn.credentials.provider | optional | org.apache.hadoop.fs.auth.DlcCloudCredentialsProvider | String | cosn 协议使用的凭证提供类。 |
| catalog.properties.io-impl | optional | (none) | String | 自定义 FileIO 实现类的全限定名。未配置时使用 HadoopFileIO， 通过作业运行环境的 Hadoop 配置读写 COS 与 CHDFS。 |
| catalog.properties.clients | optional | 2 | Integer | 访问 DLC metastore 的连接池大小。 |
| catalog.properties.list-all-tables | optional | false | Boolean | 列举表时是否包含非 Iceberg 表。 |
| partition.key | optional | (空) | String | 为目标表指定分区键，详见「分区表」章节。 |
| sink.compaction.enabled | optional | false | Boolean | 是否开启小文件合并。 |
| sink.compaction.commit.interval | optional | 10 | Integer | 每张表合并小文件的提交间隔。 |
| sink.compaction.commit.parallelism | optional | -1 | Integer | 小文件合并的并行度。取 -1 时与 Sink 写入算子的并行度一致。 |
| job.id.prefix | optional | cdc | String | 作业 ID 的前缀，用于区分不同作业。 |
| catalog.properties.* | optional | (none) | String | 透传给 catalog 的参数，属性名去掉该前缀后传给 DLC catalog。可用于配置上表未列出的 DLC、COSN 或 CHDFS 相关参数。 |
| table.properties.* | optional | (none) | String | 建表时写入 Iceberg 表属性的参数。 |
六、分区表
通过 partition.key 可以为目标表指定分区键。多条配置之间用英文分号（;）分隔，
表名与分区键之间用英文冒号（:）分隔，同一张表的多个分区键之间用英文逗号（,）分隔。
sink: type: iceberg-dlc partition.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: iceberg-dlc partition.key: testdb.table1:truncate[10](id);testdb.table2:hour(create_time);testdb.table3:day(create_time)
需要注意：
·表名需要与上游表的标识一致，格式为 namespace.tableName。
·格式不符合要求时作业启动失败。
·配置了 partition.key 时，优先使用这里的分区键；未配置时使用上游表结构自带的分区键。
七、自动建表
上游表不存在时，连接器会自动创建对应的 Iceberg 表，规则如下：
·命名空间不存在时自动创建。
·DLC 只支持单级命名空间（即 database 一层），多级命名空间会导致作业失败。
·列的类型按下表「数据类型映射」转换，并保留列注释。
·上游表的主键会被记录为 Iceberg 表的 identifier fields。
·table.properties.* 指定的参数会写入表属性，对所有新建表生效。
·列的默认值仅在表格式为 v3 及以上时写入，v2 及以下会被忽略。
·数据文件格式为 Parquet。
八、表结构变更同步
支持下列表结构变更：
| 变更类型 | 说明 |
| 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 排除对应事件类型。
九、数据类型映射
| 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 | – |
未在上表中出现的字段类型暂不支持，同步包含此类字段的表会导致作业失败。