Iceberg数据摄入连接器-实时计算 Flink版(Flink)-阿里云帮助中心

本文为您介绍如何在数据摄入 YAML 作业中，使用 Iceberg 连接器作为目标端进行数据同步。

特色功能

自动建表

写入前无需在Iceberg中预先创建目标表，数据摄入作业会根据上游的表结构自动创建对应的Iceberg表。如果下游Iceberg表已经存在，会优先使用已有的表结构进行写入，不会尝试重复建表。
表结构变更同步

支持将上游的建表、新增列、修改列类型、重命名列、删除列、清空表和删除表等事件同步到下游Iceberg表。详情请参见表结构变更。
分区字段隐式转换

通过partition.key配置分区字段时，可以直接在分区字段上使用隐式转换函数，例如truncate[10](id)、hour(create_time)、day(create_time)、month(create_time)、year(create_time)和bucket[10](create_time)。
复用已有Catalog

自VVR 11.5版本起，可以直接引用“数据管理”页面中创建的内置Iceberg Catalog，减少手写连接属性的工作量。详情请参见复用已有 Catalog。

使用限制

数据摄入YAML作业仅支持将Iceberg作为目标端，不支持作为数据源。
目标端仅支持对接阿里云数据湖构建（DLF）的Iceberg REST Catalog，catalog.properties.type固定为rest，catalog.properties.io-impl固定为org.apache.iceberg.rest.DlfFileIO。
进行ALTER COLUMN TYPE变更时，不支持修改主键列的类型。

数据摄入

Iceberg连接器可以用于数据摄入YAML作业开发，作为目标端写入。

语法结构

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.rest.signing-region: cn-beijing
  catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
  catalog.properties.warehouse: flink_iceberg
  catalog.properties.type: rest
  catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

配置项

参数	说明	是否必填	数据类型	默认值	备注
type	连接器类型。	是	STRING	无	固定值为`iceberg`。
name	目标端名称。	否	STRING	无	Sink的名称。
catalog.properties.rest.signing-region	DLF的Region ID，详见地域与服务接入点。	是	STRING	无	无
catalog.properties.uri	访问DLF Rest Catalog的URI，详见Iceberg REST。	是	STRING	无	无
catalog.properties.warehouse	DLF Catalog名称。	是	STRING	无	无
catalog.properties.warehouse	文件存储的根目录。	否	STRING	无	无
catalog.properties.type	Catalog类型，固定为rest。	是	STRING	rest	无
catalog.properties.io-impl	固定值：org.apache.iceberg.rest.DlfFileIO。	是	STRING	org.apache.iceberg.rest.DlfFileIO	无
partition.key	每个分区表的分区字段。	否	STRING	无	每个分区表的分区键，允许为多个表设置多个主键。表之间用`;`分隔，分区键之间用`,`分隔。例如，我们可以通过 `testdb.table1:id1,id2;testdb.table2:name` 来设置`testdb.table1`表的分区字段为`id1`，`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)`。
table.properties.*	创建Iceberg table的参数。	否	String	无	详情请参见Iceberg table options。

复用已有 Catalog

自VVR 11.5版本起，您可以在Flink CDC数据摄入作业中直接引用“数据管理”页面中创建的内置Iceberg Catalog，减少手写连接属性工作量。

sink:
  type: iceberg
  using.built-in-catalog: iceberg_catalog

目前，数据摄入作业支持自动复用所有 Iceberg Catalog 参数，等价于在 YAML 作业中手动配置 catalog.properties.前缀的参数。

如果希望覆盖以上自动复用的参数，可显式写出相应的 YAML 参数，其具备更高的优先级。

使用示例

Iceberg Catalog为DLF Catalog，写入阿里云数据湖构建的配置示例：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.rest.signing-region: cn-beijing
  catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
  catalog.properties.warehouse: flink_iceberg
  catalog.properties.type: rest
  catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

其中，catalog.properties前缀的参数含义请参见创建Iceberg DLF Catalog。

表结构变更

目前，Iceberg作为数据摄入目标端支持以下表结构变更事件：

CREATE TABLE EVENT
ADD COLUMN EVENT
ALTER COLUMN TYPE EVENT（不支持修改主键列的类型）
RENAME COLUMN EVENT
DROP COLUMN EVENT
TRUNCATE TABLE EVENT
DROP TABLE EVENT

说明

在下游 Iceberg 表已经存在时，会优先使用已有的表结构进行写入，不会尝试重复建表。