Iceberg Pipeline Connector

The Iceberg Pipeline Connector functions as a Data Sink for data pipelines, enabling data writes to Apache Iceberg tables. This document explains how to configure the connector.

Key Capabilities

• Automatic Table Creation: creates Iceberg tables dynamically when they do not exist
• Schema Synchronization: propagates schema changes (e.g., column additions) from source systems to Iceberg
• Data Replication: supports both batch and streaming data synchronization

How to create Pipeline

The pipeline for reading data from MySQL and sink to Iceberg can be defined as follows:

Hadoop Catalog Example

source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
  server-id: 5401-5404

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.type: hadoop
  catalog.properties.warehouse: /path/warehouse

pipeline:
  name: MySQL to Iceberg Pipeline
  parallelism: 2

AWS Glue Catalog Example

source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
  server-id: 5401-5404

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.type: glue
  catalog.properties.warehouse: s3://my-bucket/warehouse
  catalog.properties.io-impl: org.apache.iceberg.aws.s3.S3FileIO
  catalog.properties.client.region: us-east-1
  catalog.properties.glue.skip-archive: true

pipeline:
  name: MySQL to Iceberg via Glue Pipeline
  parallelism: 2

Note: Depending on the catalog type, you may need to add extra JARs manually and pass them with the --jar argument of Flink CDC CLI when submitting YAML pipeline jobs.

Catalog Type	Dependency Item	Description
all	org.apache.iceberg:iceberg-flink-runtime-1.20	Iceberg Flink runtime. Required for all catalog types when not pre-installed in the runtime environment (e.g., standalone Flink clusters).
hadoop	org.apache.flink:flink-shaded-hadoop-2-uber:2.8.3-10.0	Provides Hadoop filesystem dependencies.
glue	org.apache.iceberg:iceberg-aws	Provides AWS Glue Catalog and S3 FileIO implementation.
glue	software.amazon.awssdk:bundle	AWS SDK bundle required by iceberg-aws.

Pipeline Connector Options

Option	Required	Default	Type	Description
type	required	(none)	String	Specify what connector to use, here should be iceberg.
name	optional	(none)	String	The name of the sink.
catalog.properties.type	conditionally required	(none)	String	Metastore type of Iceberg catalog, supports hadoop, hive, and glue. Either this option or catalog.properties.catalog-impl must be set.
catalog.properties.catalog-impl	conditionally required	(none)	String	Custom catalog implementation class, e.g. org.apache.iceberg.aws.glue.GlueCatalog. Either this option or catalog.properties.type must be set.
catalog.properties.warehouse	optional	(none)	String	The warehouse root path of the Iceberg catalog, used by all catalog types. For hadoop and hive catalogs, this is typically a local or distributed filesystem path. For glue catalog, this is typically an object storage path like s3://my-bucket/warehouse.
catalog.properties.uri	optional	(none)	String	URI of the metastore server (e.g. Hive Metastore thrift URI).
catalog.properties.io-impl	optional	(none)	String	Custom FileIO implementation class. For AWS S3, use org.apache.iceberg.aws.s3.S3FileIO.
catalog.properties.client.region	optional	(none)	String	The AWS region for the Glue catalog client (e.g. us-east-1).
catalog.properties.glue.id	optional	(none)	String	The Glue catalog ID (AWS account ID). By default, the caller's AWS account ID is used.
catalog.properties.glue.skip-archive	optional	true	Boolean	Whether to skip archiving older table versions in Glue.
catalog.properties.glue.skip-name-validation	optional	false	Boolean	Whether to skip name validation for Glue catalog.
partition.key	optional	(none)	String	Partition keys for each partitioned table. Allow setting multiple primary keys for multiTables. Tables are separated by ';', and partition keys are separated by ','. For example, we can set partition.key of two tables using 'testdb.table1:id1,id2;testdb.table2:name'. For partition transforms, we can set partition.key using '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)'.
catalog.properties.*	optional	(none)	String	Pass Iceberg catalog options to the pipeline. See Iceberg catalog options.
table.properties.*	optional	(none)	String	Pass Iceberg table options to the pipeline. See Iceberg table options.
hadoop.conf.*	optional	(none)	String	Pass Hadoop Configuration options used by Iceberg catalog/table operations. The prefix hadoop.conf. will be stripped. For example, hadoop.conf.fs.s3a.endpoint.

Usage Notes

• The source table must have a primary key. Tables with no primary key are not supported.

• Exactly-once semantics are not supported. The connector uses at-least-once + the table's primary key for idempotent writing.

Data Type Mapping

CDC type	Iceberg type	NOTE
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)
BOOLEAN	BOOLEAN
DATE	DATE
TIMESTAMP	TIMESTAMP
TIMESTAMP_LTZ	TIMESTAMP_LTZ
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)