DLC 连接器可以作为 Sink（
type: iceberg-dlc），把上游的变更事件写入腾讯云数据湖计算（Data Lake Compute，DLC）托管的 Iceberg 表，提供自动建表、表结构变更同步与数据实时同步能力。数据以 Apache Iceberg 格式写入，由 DLC 托管的 catalog 管理元数据。
前置准备
已开通腾讯云 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: mysqlname: MySQL Sourcehostname: 127.0.0.1port: 3306username: adminpassword: passtables: adb.\\.*server-id: 5401-5404sink:type: iceberg-dlcname: DLC Sinkcatalog.properties.uri: dlc.internal.tencentcloudapi.comcatalog.properties.qcloud.dlc.region: ap-guangzhoucatalog.properties.qcloud.dlc.secret-id: your-secret-idcatalog.properties.qcloud.dlc.secret-key: your-secret-keycatalog.properties.qcloud.dlc.user.appid: your-appidpipeline:name: MySQL to DLC Pipelineparallelism: 2
不配置
catalog.properties.warehouse 时，使用 DLC 托管表，表数据的存储位置由 DLC 服务端分配。
如果需要自行指定存储位置，可以配置
catalog.properties.warehouse，取值必须使用
cosn:// 或 lakefs:// 协议：sink:type: iceberg-dlcname: DLC Sinkcatalog.properties.qcloud.dlc.region: ap-guangzhoucatalog.properties.qcloud.dlc.secret-id: your-secret-idcatalog.properties.qcloud.dlc.secret-key: your-secret-keycatalog.properties.qcloud.dlc.user.appid: your-appidcatalog.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-dlcpartition.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-dlcpartition.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 | - |
未在上表中出现的字段类型暂不支持，同步包含此类字段的表会导致作业失败。
