Documentation
Home Page Realtime Compute for Apache Flink Streaming Compute Flink Developer Reference Supported connectors Apache Iceberg connector

Apache Iceberg connector

Updated at:
Copy as MD
Product

This topic describes how to use the Apache Iceberg connector.

Background information

Apache Iceberg is an open data lake table format. You can use Apache Iceberg to quickly build your own data lake storage service on HDFS or cloud OSS, and use computing engines from the open-source big data ecosystem, such as Flink, Spark, Hive, and Presto, to perform data lake analysis.

Category

Description

Supported types

Source table, sink table, and data ingestion sink

Running modes

Batch mode and streaming mode

Data format

Not applicable

Specific metrics

None

API types

SQL, data ingestion YAML job

Supports updating or deleting data in sink tables

Yes

Features

The Apache Iceberg connector provides the following features:

  • Builds a low-cost, lightweight data lake storage service based on HDFS or object storage.

  • Provides full ACID semantics.

  • Supports time travel queries to access historical data versions.

  • Supports efficient data filtering.

  • Supports schema evolution.

  • Supports partition evolution.

Note

Use Flink's fault tolerance and stream processing capabilities to import large volumes of log data into an Apache Iceberg data lake in real time. You can then use Flink or other analysis engines to extract value from the data.

Limits

  • The Apache Iceberg connector is supported only in Realtime Compute for Apache Flink with Ververica Runtime (VVR) 4.0.8 or later. The Apache Iceberg connector must be used with a Data Lake Formation (DLF) catalog. For more information, see Manage DLF-Legacy catalogs.

  • The Apache Iceberg connector supports Apache Iceberg v1 and v2 table formats. For more information, see the Iceberg Table Spec.

    Note

    Only Realtime Compute for Apache Flink that uses VVR 8.0.7 or later supports the v2 table format.

  • In streaming read mode, you can use only append-only Iceberg tables as source tables.

Syntax

CREATE TABLE iceberg_table (
  id    BIGINT,
  data  STRING
  PRIMARY KEY(`id`) NOT ENFORCED
)
 PARTITIONED BY (data)
 WITH (
 'connector' = 'iceberg',
  ...
);

WITH options

Common options

Parameter

Description

Type

Required

Default

Remarks

connector

The type of the connector.

String

Yes

None

The value must beiceberg.

catalog-name

The name of the catalog.

String

Yes

None

Enter a custom name in English.

catalog-database

The name of the database.

String

Yes

default

The name of your database in Data Lake Formation (DLF), such as dlf_db.

Note

If you do not have a Data Lake Formation (DLF) database, create one first.

io-impl

The implementation class for the distributed file system.

String

Yes

None

The value must beorg.apache.iceberg.aliyun.oss.OSSFileIO.

oss.endpoint

The endpoint of Alibaba Cloud Object Storage Service (OSS).

String

No

None

For more information, see Regions and Endpoints.

Note

  • We recommend that you set the oss.endpoint parameter to the VPC endpoint for OSS. For example, if you select the China (Hangzhou) region, set the oss.endpoint parameter to oss-cn-hangzhou-internal.aliyuncs.com.

  • If you need to access OSS across VPCs, see How do I access other services across VPCs?

  • access.key.id: for VVR 8.0.6 and earlier

  • access-key-id: for VVR 8.0.7 and later

Your Alibaba Cloud account AccessKey ID.

String

Yes

None

For more information, see How do I view the AccessKey ID and AccessKey secret?

Important

To prevent your AccessKey information from being leaked, we recommend that you use variables to specify the AccessKey values. For more information, see Project variables.

  • access.key.secret: for VVR 8.0.6 and earlier

  • access-key-secret: for VVR 8.0.7 and later

Your Alibaba Cloud account AccessKey secret.

String

Yes

None

catalog-impl

The class name of the catalog.

String

Yes

None

The value must beorg.apache.iceberg.aliyun.dlf.DlfCatalog.

warehouse

The OSS path to store table data.

String

Yes

None

None

dlf.catalog-id

Your Alibaba Cloud account ID.

String

Yes

None

You can obtain the account ID on the User Information page.

dlf.endpoint

The endpoint of Data Lake Formation (DLF).

String

Yes

None

.

Note

  • We recommend that you set the dlf.endpoint parameter to the VPC endpoint of DLF. For example, if you select the China (Hangzhou) region, set the dlf.endpoint parameter to dlf-vpc.cn-hangzhou.aliyuncs.com.

  • If you need to access DLF across VPCs, see Workspace management and operations

dlf.region-id

The region of Data Lake Formation (DLF).

String

Yes

None

.

Note

Make sure that the region is the same as the region specified for the dlf.endpoint parameter.

Sink-only options

Parameter

Description

Type

Required

Default

Remarks

write.operation

The write operation mode.

String

No

upsert

  • upsert (default): Updates data.

  • insert: Appends data.

  • bulk_insert: Performs a bulk insert without updating existing data.

hive_sync.enable

Specifies whether to synchronize metadata to Hive.

Boolean

No

false

Valid values:

  • true: Enables synchronization.

  • false (default): Disables synchronization.

hive_sync.mode

The Hive metadata synchronization mode.

String

No

hms

  • hms (default): Set this value if you use a DLF catalog.

  • jdbc: Set this value if you use a JDBC catalog.

hive_sync.db

The name of the Hive database to which data is synchronized.

String

No

The database name of the current table in the catalog.

None

hive_sync.table

The name of the Hive table to which data is synchronized.

String

No

The name of the current table.

None

dlf.catalog.region

The region of Data Lake Formation (DLF).

String

No

None

.

Note

  • The dlf.catalog.region parameter takes effect only when the hive_sync.mode parameter is set tohms.

  • Make sure that the region is the same as the region specified for the dlf.catalog.endpoint parameter.

dlf.catalog.endpoint

The endpoint of Data Lake Formation (DLF).

String

No

None

.

Note

  • The dlf.catalog.endpoint parameter takes effect only when the hive_sync.mode parameter is set to hms.

  • We recommend that you set the dlf.catalog.endpoint parameter to the VPC endpoint of DLF. For example, if you select the China (Hangzhou) region, set the dlf.catalog.endpoint parameter to dlf-vpc.cn-hangzhou.aliyuncs.com.

  • If you need to access DLF across VPCs, see Workspace management and operations

Data type mapping

Iceberg type

Flink type

BOOLEAN

BOOLEAN

INT

INT

LONG

BIGINT

FLOAT

FLOAT

DOUBLE

DOUBLE

DECIMAL(P,S)

DECIMAL(P,S)

DATE

DATE

TIME

TIME

Note

Iceberg timestamps have microsecond precision, whereas Flink timestamps have millisecond precision. When you use Flink to read data from Iceberg, Flink converts the time precision to milliseconds.

TIMESTAMP

TIMESTAMP

TIMESTAMPTZ

TIMESTAMP_LTZ

STRING

STRING

FIXED(L)

BYTES

BINARY

VARBINARY

STRUCT<...>

ROW

LIST<E>

LIST

MAP<K,V>

MAP

Examples

Ensure that you have an OSS bucket and a Data Lake Formation (DLF) database. For more information, see Create a bucket and Databases, tables, and functions.

Note

When you specify a path for your Data Lake Formation (DLF) database, we recommend that you follow the ${warehouse}/${database_name}.db format. For example, if the warehouse address is oss://iceberg-test/warehouse and the database name is dlf_db, set the OSS path of dlf_db to oss://iceberg-test/warehouse/dlf_db.db.

Sink table example

This example uses the Datagen connector to generate random streaming data and write the data to an Iceberg table.

CREATE TEMPORARY TABLE datagen(
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE dlf_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

INSERT INTO dlf_iceberg SELECT * FROM datagen;

Source table examples

  • Use a DLF catalog to write data from an Iceberg source table to an Iceberg sink table.

    CREATE TEMPORARY TABLE src_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

CREATE TEMPORARY TABLE dst_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

BEGIN STATEMENT SET;

INSERT INTO src_iceberg VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC'), (4, 'DDD'), (5, 'EEE');
INSERT INTO dst_iceberg SELECT * FROM src_iceberg;

END;

Data ingestion

You can use the Apache Iceberg connector as a sink in a YAML job for data ingestion.

Syntax

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

Parameters

Parameter

Description

Required

Type

Default

Remarks

type

The type of the connector.

Yes

STRING

None

The fixed value is iceberg.

name

The name of the sink.

No

STRING

None

The name of the sink.

catalog.properties.rest.signing-region

The region ID of DLF. For more information, see Service endpoints.

Yes

STRING

None

None

catalog.properties.uri

The URI used to access the DLF REST catalog. For more information, see Iceberg REST.

Yes

STRING

None

None

catalog.properties.warehouse

The root directory for file storage.

Yes

STRING

None

None

catalog.properties.warehouse

The root directory for file storage.

No

STRING

None

None

catalog.properties.type

The type of the catalog. The value must be rest.

Yes

STRING

rest

None

catalog.properties.io-impl

The value must be org.apache.iceberg.rest.DlfFileIO.

Yes

STRING

org.apache.iceberg.rest.DlfFileIO

None

partition.key

The partition key for each partitioned table.

No

STRING

None

You can set partition keys for multiple tables. Separate table definitions with a semicolon (;) and partition keys with a comma (,). For example, you can specify testdb.table1:id1,id2;testdb.table2:name to set the partition keys for the testdb.table1 table to id1 and id2, and the partition key for the testdb.table2 table to name.

For partitions that require implicit transformations, add the transformation function directly to the partition field. Example: 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.*

The parameters for creating an Iceberg table.

No

String

None

For more information, see Iceberg table options.

Reuse an existing catalog

Starting with VVR 11.5, you can directly reference a built-in Iceberg catalog that is created on the Data Management page in a Flink CDC data ingestion job. This simplifies your configuration by reducing the number of required connection properties.

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

Data ingestion jobs can automatically reuse all Iceberg catalog parameters. This is equivalent to manually configuring parameters with the catalog.properties. prefix in your YAML job.

If you want to override the reused parameters, you can explicitly specify the corresponding YAML parameters. These parameters take precedence.

Example

The following example shows how to use a DLF catalog as an Iceberg catalog and write data to Data Lake Formation (DLF):

  • 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

    For information about the parameters that are prefixed with catalog.properties, see Create an Iceberg DLF catalog.

Schema changes

When used as a data ingestion sink, the Apache Iceberg connector supports the following schema changes:

  • CREATE TABLE

  • ADD COLUMN

  • ALTER COLUMN TYPE (Modifying the type of a primary key column is not supported.)

  • RENAME COLUMN

  • DROP COLUMN

  • TRUNCATE TABLE

  • DROP TABLE

Note

If the downstream Iceberg table already exists, the job uses the existing table schema for writes and does not create the table again.

Related documentation

For more information about connectors that are supported by Flink, see Supported connectors.

Previous：PolarDB-X CDC (Public preview)Next：Faker connector