Realtime Compute for Apache Flink:Apache Iceberg

DLF Catalog parameters (recommended)

Use these parameters when connecting to an Iceberg table managed by a DLF Catalog.

Hive Catalog parameters

Use these parameters when connecting to an Iceberg table managed by a self-managed Hive Metastore (HMS). Make sure the Flink cluster can reach the HMS cluster over the network.

Sink-specific parameters

Prerequisites

Before running the examples, complete the following:

• An OSS bucket. See Create a bucket in the console.

• A DLF database. See Databases, tables, and functions.

When creating the DLF database, set the Path using the format ${warehouse}/${database_name}.db. For example, if your warehouse is oss://iceberg-test/warehouse and the database name is dlf_db, set the path to oss://iceberg-test/warehouse/dlf_db.db.

Write streaming data to an Iceberg sink table

This example uses the Datagen connector to generate random streaming data and write it to an Iceberg table via a DLF Catalog.

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;

Read from an Iceberg source table (Hive Catalog)

This example reads from an Iceberg table managed by a self-managed Hive Metastore. The table data is stored at oss://<bucket>/<path>/<database-name>/flink_table.

CREATE TEMPORARY TABLE flink_table (
  id   BIGINT,
  data STRING
) WITH (
  'connector'        = 'iceberg',
  'catalog-name'     = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'uri'              = 'thrift://<ip>:<port>',
  'warehouse'        = 'oss://<bucket>/<path>',
  'io-impl'          = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'access-key-id'    = '<yourAccessKeyId>',
  'access-key-secret'= '<yourAccessKeySecret>',
  'oss.endpoint'     = '<yourOSSEndpoint>'
);

Copy data between two Iceberg tables (DLF Catalog)

This example inserts seed data into a source Iceberg table and copies it to a destination table, both using a DLF Catalog.

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;

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

Configuration parameters

Example: MySQL to Iceberg via DLF REST Catalog

The following example ingests data from MySQL into Data Lake Formation using a DLF REST Catalog. For all catalog.properties.* parameters, see Create an Iceberg 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

Schema evolution

When using Iceberg as a data ingestion sink, the following schema change events are automatically propagated:

• 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

If the downstream Iceberg table already exists, data is written using the existing table schema. The system does not attempt to recreate the table.