Skip to main content
Version: Next

Apache Iceberg

Apache Iceberg sink connector

Support Iceberg Version​

  • 1.10.2, writing tables of format version 1, 2 and 3

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Description​

Sink connector for Apache Iceberg. It supports CDC writes, automatic table creation, table schema evolution, and multi-table write jobs.

Key features​

Supported DataSource Info​

DatasourceDependentMaven
Iceberghive-execDownload
Iceberglibfb303Download

Database Dependency​

In order to be compatible with different versions of Hadoop and Hive, the scope of hive-exec in the project pom file are provided, so if you use the Flink engine, first you may need to add the following Jar packages to <FLINK_HOME>/lib directory, if you are using the Spark engine and integrated with Hadoop, then you do not need to add the following Jar packages.

hive-exec-xxx.jar
libfb303-xxx.jar

Some versions of the hive-exec package do not have libfb303-xxx.jar, so you also need to manually import the Jar package.

Data Type Mapping​

SeaTunnel Data typeIceberg Data type
BOOLEANBOOLEAN
INTINTEGER
BIGINTLONG
FLOATFLOAT
DOUBLEDOUBLE
DATEDATE
TIMETIME
TIMESTAMPTIMESTAMP
TIMESTAMP_TZTIMESTAMPTZ
STRINGSTRING
BYTESFIXED
BINARY
DECIMALDECIMAL
ROWSTRUCT
ARRAYLIST
MAPMAP

When the sink creates a table with format-version = 3, TIMESTAMP and TIMESTAMP_TZ columns with nanosecond precision (column scale 9, for example an Iceberg timestamp_ns column read by the Iceberg source) are created as TIMESTAMP_NS and TIMESTAMPTZ_NS. These types only hold values from the year 1677 to 2262; a value outside that range fails the write. Other scales, nested fields, columns added by schema evolution and tables of format version 1 and 2 keep microsecond timestamps. On the Spark engine, rows carry TIMESTAMP values in microseconds and TIMESTAMP_TZ values in milliseconds, so the extra precision is already cut before it reaches the sink.

Writing into an existing format version 3 table also accepts TIMESTAMP_NS and TIMESTAMPTZ_NS columns. UNKNOWN columns only accept null values. VARIANT columns are not supported yet: leave them out of the written rows, which works for Parquet data files only. Tables with GEOMETRY or GEOGRAPHY columns cannot be written yet.

Sink Options​

NameTypeRequiredDefaultDescription
catalog_namestringnodefaultUser-specified catalog name. default is default
namespacestringnodefaultThe iceberg database name in the backend catalog. default is default
tablestringno-The iceberg table name in the backend catalog. If not set, the table name of the upstream table is used.
iceberg.catalog.configmapyes-Specify the properties for initializing the Iceberg catalog, which can be referenced in this file: CatalogProperties.java
hadoop.configmapno-Properties passed through to the Hadoop configuration
iceberg.hadoop-conf-pathstringno-The specified loading paths for the 'core-site.xml', 'hdfs-site.xml', 'hive-site.xml' files.
case_sensitivebooleannofalseIf data columns where selected via schema [config], controls whether the match to the schema will be done with case sensitivity.
iceberg.table.write-propsmapno-Properties passed through to Iceberg writer initialization, these take precedence, such as 'write.format.default', 'write.target-file-size-bytes', and other settings, can be found with specific parameters at TableProperties.java.
iceberg.table.auto-create-propsmapno-Configuration specified by Iceberg during automatic table creation.
iceberg.table.schema-evolution-enabledbooleannofalseSetting to true enables Iceberg tables to support schema evolution during the synchronization process
iceberg.table.primary-keysstringno-Default comma-separated list of columns that identify a row in tables (primary key)
iceberg.table.partition-keysstringno-Default comma-separated list of partition fields to use when creating tables. Supports placeholder ${partition_keys} for multi-table jobs
iceberg.table.upsert-mode-enabledbooleannofalseSet to true to enable upsert mode, default is false
schema_save_modeEnumnoCREATE_SCHEMA_WHEN_NOT_EXISTthe schema save mode, please refer to schema_save_mode below
data_save_modeEnumnoAPPEND_DATAthe data save mode, please refer to data_save_mode below
custom_sqlstringRequired when data_save_mode is CUSTOM_PROCESSING-Custom delete SQL for CUSTOM_PROCESSING data save mode, for example delete from ... where ....
iceberg.table.commit-branchstringno-Default branch for commits
multi_table_sink_replicaintno1Replica number for every table writer in multi-table sink mode. Use this when one upstream job writes to multiple Iceberg tables and each table needs more than one sink writer.
krb5_pathstringno/etc/krb5.confThe path of krb5.conf, used for Kerberos authentication.
kerberos_principalstringno-The principal for Kerberos authentication.
kerberos_keytab_pathstringno-The keytab file path for Kerberos authentication.

Sink Option descriptions​

iceberg.table.auto-create-props [map]​

Table properties used when the sink creates the table. Set format-version to choose the Iceberg table format version; supported values are 1, 2 and 3, and Iceberg creates format version 2 tables when it is not set. Any other value fails the job before the table is created. The property only applies to tables the sink creates: an existing table keeps its format version, upgrade it with your Iceberg engine, for example ALTER TABLE ... SET TBLPROPERTIES ('format-version' = '3') in Spark.

iceberg.table.auto-create-props = {
  "format-version" = "3"
}

Format version 3 tables support append writes. Upsert and CDC writes are rejected with error ICEBERG-04 before any data is written, because the sink writes position delete files, which format version 3 replaces with deletion vectors. This covers iceberg.table.primary-keys, iceberg.table.upsert-mode-enabled and existing tables with identifier fields, and also creating a format version 3 table for a source table with a primary key, which would otherwise become the identifier fields. Use a format version 2 table for upsert and CDC pipelines.

iceberg.table.upsert-mode-enabled [boolean]​

When this option is true, iceberg.table.primary-keys must be configured explicitly. The sink does not inherit primary keys from the source table automatically.

iceberg.table.partition-keys [string]​

Use a comma-separated list, for example dt,region or Iceberg transforms such as days(ts). In multi-table jobs, ${partition_keys} can be used as a placeholder from the upstream table metadata.

iceberg.table.commit-branch [string]​

Write commits to the specified Iceberg branch. Leave it empty to commit to the table's default branch.

custom_sql [string]​

When data_save_mode = CUSTOM_PROCESSING, configure the delete SQL that removes the target data before the sink writes. This option is required in that mode.

For Iceberg timestamptz columns, comparison and IN/NOT IN predicates accept timestamp literals with an offset, for example event_time >= '2026-09-12 10:00:00.123456+05:30'. Offsets are converted to the corresponding UTC instant at microsecond precision. Offset-free literals retain their existing UTC interpretation. Iceberg timestamp columns without a time zone continue to require offset-free literals.

The existing file-level delete restriction still applies: Iceberg rejects a delete when only some rows in a data file match the condition.

krb5_path [string]​

The path of krb5.conf, used for Kerberos authentication.

kerberos_principal [string]​

The principal for Kerberos authentication.

kerberos_keytab_path [string]​

The keytab file path for Kerberos authentication.

Task Example​

Simple​

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  MySQL-CDC {
    plugin_output = "customers_mysql_cdc_iceberg"
    server-id = 5652
    username = "st_user"
    password = "seatunnel"
    table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
    url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
  }
}

transform {
}

sink {
  Iceberg {
    catalog_name="seatunnel_test"
    iceberg.catalog.config={
      "type"="hadoop"
      "warehouse"="file:///tmp/seatunnel/iceberg/hadoop-sink/"
    }
    namespace="seatunnel_namespace"
    table="iceberg_sink_table"
    iceberg.table.write-props={
      write.format.default="parquet"
      write.target-file-size-bytes=536870912
    }
    iceberg.table.primary-keys="id"
    iceberg.table.partition-keys="f_datetime"
    iceberg.table.upsert-mode-enabled=true
    iceberg.table.schema-evolution-enabled=true
    case_sensitive=true
  }
}

Hive Catalog​

sink {
  Iceberg {
    catalog_name = "seatunnel_test"
    iceberg.catalog.config = {
      type = "hive"
      uri = "thrift://localhost:9083"
      warehouse = "hdfs://your_cluster/tmp/seatunnel/iceberg/"
    }
    namespace = "seatunnel_namespace"
    table = "iceberg_sink_table"
    iceberg.table.write-props = {
      write.format.default = "parquet"
      write.target-file-size-bytes = 536870912
    }
    iceberg.table.primary-keys = "id"
    iceberg.table.partition-keys = "f_datetime"
    iceberg.table.upsert-mode-enabled = true
    iceberg.table.schema-evolution-enabled = true
    case_sensitive = true
  }
}

Hadoop Catalog​

sink {
  Iceberg {
    catalog_name = "seatunnel_test"
    iceberg.catalog.config = {
      type = "hadoop"
      warehouse = "hdfs://your_cluster/tmp/seatunnel/iceberg/"
    }
    namespace = "seatunnel_namespace"
    table = "iceberg_sink_table"
    iceberg.table.write-props = {
      write.format.default = "parquet"
      write.target-file-size-bytes = 536870912
    }
    iceberg.table.primary-keys = "id"
    iceberg.table.partition-keys = "f_datetime"
    iceberg.table.upsert-mode-enabled = true
    iceberg.table.schema-evolution-enabled = true
    case_sensitive = true
  }
}

Commit To Iceberg Branch​

sink {
  Iceberg {
    catalog_name = "seatunnel_test"
    iceberg.catalog.config = {
      type = "hadoop"
      warehouse = "file:///tmp/seatunnel/iceberg/hadoop-sink/"
    }
    namespace = "seatunnel_namespace"
    table = "iceberg_sink_table"
    iceberg.table.commit-branch = "audit_branch"
  }
}

Glue Catalog​

sink {
  Iceberg {
    catalog_name = "seatunnel_test"
    iceberg.catalog.config = {
      warehouse     = "s3://your-bucket/warehouse/"
      catalog-impl  = "org.apache.iceberg.aws.glue.GlueCatalog"
      io-impl       = "org.apache.iceberg.aws.s3.S3FileIO"
      client.region = "your-region"
    }
    namespace = "seatunnel_namespace"
    table     = "iceberg_sink_table"
    iceberg.table.write-props = {
      write.format.default = "parquet"
      write.target-file-size-bytes = 536870912
    }
    iceberg.table.primary-keys = "id"
    iceberg.table.partition-keys = "f_datetime"
    iceberg.table.upsert-mode-enabled = true
    iceberg.table.schema-evolution-enabled = true
    case_sensitive = true
  }
}

AWS S3 Tables REST Catalog​

Amazon S3 Tables is a storage service for tabular data that's optimized for analytics workloads, with features designed to continuously improve query performance and reduce storage costs for tables. S3 Tables is purpose-built for storing tabular data, such as daily purchase transactions, streaming sensor data, or ad impressions. Tabular data represents data in columns and rows, like in a database table.

You can connect an Iceberg REST client to the Amazon S3 Tables Iceberg REST endpoint and then make REST API calls to create, update, or query tables in S3 table buckets. The endpoint implements a standardized set of Iceberg REST APIs specified in the Apache Iceberg REST Catalog Open API specification. The endpoint works by translating Iceberg REST API operations to corresponding S3 Tables operations.

Data in S3 Tables is stored in a new bucket type: table buckets, which store tables as subresources. Table buckets support storing tables in Apache Iceberg format. Using standard SQL statements, you can query tables through Iceberg-compatible query engines such as Amazon Athena, Amazon Redshift, and Apache Spark.

sink {
  Iceberg {
    catalog_name = "s3_tables_catalog"
    namespace = "s3_tables_catalog"
    table = "user_data"

    iceberg.catalog.config = {
      type: "rest"
      warehouse: "arn:aws:s3tables:<Region>:<accountID>:bucket/<bucketname>"
      uri: "https://s3tables.<Region>.amazonaws.com/iceberg"
      rest.sigv4-enabled: "true"
      rest.signing-name: "s3tables"
      rest.signing-region: "<Region>"
    }
  }
}

Kerberos Authentication​

The following example demonstrates how to configure Iceberg sink with Kerberos authentication when using Hadoop catalog with HDFS:

sink {
  Iceberg {
    catalog_name = "seatunnel_test"
    iceberg.catalog.config = {
      type = "hadoop"
      warehouse = "hdfs://your_cluster/tmp/seatunnel/iceberg/"
    }
    namespace = "seatunnel_namespace"
    table = "iceberg_sink_table"
    iceberg.table.write-props = {
      write.format.default = "parquet"
      write.target-file-size-bytes = 536870912
    }
    krb5_path = "/etc/krb5.conf"
    kerberos_principal = "hive/your_host@EXAMPLE.COM"
    kerberos_keytab_path = "/path/to/your.keytab"
    iceberg.table.primary-keys = "id"
    iceberg.table.partition-keys = "f_datetime"
    iceberg.table.upsert-mode-enabled = true
    iceberg.table.schema-evolution-enabled = true
    case_sensitive = true
  }
}

Description:

  • krb5_path: The path to the krb5.conf file used for Kerberos authentication.
  • kerberos_principal: The principal for Kerberos authentication in the format primary/instance@REALM.
  • kerberos_keytab_path: The keytab file path for Kerberos authentication.

Multiple table​

example1​

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  Mysql-CDC {
    url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
    username = "root"
    password = "******"
    
    table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]
  }
}

transform {
}

sink {
  Iceberg {
    ...
    namespace = "${database_name}_test"
    table = "${table_name}_test"
  }
}

example2​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Jdbc {
    driver = oracle.jdbc.driver.OracleDriver
    url = "jdbc:oracle:thin:@localhost:1521/XE"
    user = testUser
    password = testPassword

    table_list = [
      {
        table_path = "TESTSCHEMA.TABLE_1"
      },
      {
        table_path = "TESTSCHEMA.TABLE_2"
      }
    ]
  }
}

transform {
}

sink {
  Iceberg {
    ...
    namespace = "${schema_name}_test"
    table = "${table_name}_test"
  }
}

Changelog​

Change Log
ChangeCommitVersion
[Chore] fix typos filed -> field (#9757)https://github.com/apache/seatunnel/commit/e3e1c67d292.3.12
[Improve][Core] Unify the aws-sdk-v2 version to 2.31.30 (#9698)https://github.com/apache/seatunnel/commit/41c251cc8a2.3.12
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)https://github.com/apache/seatunnel/commit/9212a771402.3.12
[Bug][Connector-V2] Fix the issue of writing the ORC format Iceberg report "Illegal provider-class name" (#6754) (#9588)https://github.com/apache/seatunnel/commit/74b193dd5a2.3.12
[Bug][Connector-V2] Updates Iceberg version to 1.6.1 (#9387) (#9451)https://github.com/apache/seatunnel/commit/7b92a6c5c12.3.12
[Fix][Connector-Iceberg] Fix Time Zone Issue for Iceberg Timestamp Type (#9460)https://github.com/apache/seatunnel/commit/60cd4976102.3.12
[Feature][Connector-V2] Iceberg add glue catalog support (#9247)https://github.com/apache/seatunnel/commit/ecff2e86182.3.11
[Improve] Remove useless iceberg sink config iceberg.table.config (#9307)https://github.com/apache/seatunnel/commit/fbdf39ebf22.3.11
[Improve][connector-iceberg] fix schema change event (#9217)https://github.com/apache/seatunnel/commit/56669095b72.3.11
[Feature][Transform] Support define sink column type (#9114)https://github.com/apache/seatunnel/commit/ab7119e5072.3.11
[Feat][Connector-v2][Iceberg]support filter conditions in iceberg source (#9095)https://github.com/apache/seatunnel/commit/0eb72780ee2.3.11
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Fix][API] Fixed not invoke the SinkAggregatedCommitter's init method (#9070)https://github.com/apache/seatunnel/commit/df0d11d6322.3.11
[Improve] iceberg options (#8967)https://github.com/apache/seatunnel/commit/82a374ec872.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Feature][Iceberg] Support read multi-table (#8524)https://github.com/apache/seatunnel/commit/2bfb97e5022.3.10
[Improve][Iceberg] Filter catalog table primaryKey is empty (#8413)https://github.com/apache/seatunnel/commit/857aab5e832.3.9
[Improve][Connector-V2] Reduce the create times of iceberg sink writer (#8155)https://github.com/apache/seatunnel/commit/45a7a715a22.3.9
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Iceberg] Support custom delete sql for sink savemode (#8094)https://github.com/apache/seatunnel/commit/29ca928c362.3.9
[Improve][Connector-V2] Reduce the request times of iceberg load table (#8149)https://github.com/apache/seatunnel/commit/555f5eb4042.3.9
[Feature][Core] Support cdc task ddl restore for zeta (#7463)https://github.com/apache/seatunnel/commit/8e322281ed2.3.9
[Improve][Iceberg] Support table comment for catalog (#7936)https://github.com/apache/seatunnel/commit/72ab38f3172.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Fix][Connector-V2] Fix iceberg throw java: package sun.security.krb5 does not exist when use jdk 11 (#7734)https://github.com/apache/seatunnel/commit/116af4febc2.3.8
[Hotfix][Connector-V2] Release resources when task is closed for iceberg sinkwriter (#7729)https://github.com/apache/seatunnel/commit/ff281183bd2.3.8
[Fix][Connector-V2] Fixed iceberg sink can not handle uppercase fields (#7660)https://github.com/apache/seatunnel/commit/b7be0cb4a12.3.8
[Hotfix][CDC] Fix ddl duplicate execution error when config multi_table_sink_replica (#7634)https://github.com/apache/seatunnel/commit/23ab3edbbb2.3.8
[Improve][Iceberg] Add savemode create table primaryKey testcase (#7641)https://github.com/apache/seatunnel/commit/6b36f90f4d2.3.8
[Hotfix] Fix iceberg missing column comment when savemode create table (#7608)https://github.com/apache/seatunnel/commit/b35bd94bfb2.3.8
[Improve][Connector-V2] Remove hard code iceberg table format version (#7500)https://github.com/apache/seatunnel/commit/f49b263e652.3.8
[Improve][API] Move catalog open to SaveModeHandler (#7439)https://github.com/apache/seatunnel/commit/8c2c5c79a12.3.8
[Feature][Connector-V2][Iceberg] Support Iceberg Kerberos (#7246)https://github.com/apache/seatunnel/commit/e3001207c82.3.8
[Improve][Connector] Add multi-table sink option check (#7360)https://github.com/apache/seatunnel/commit/2489f6446b2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)https://github.com/apache/seatunnel/commit/c4ca74122c2.3.6
[Bug][Connector-Iceberg]fix create iceberg v2 table with pks (#6895)https://github.com/apache/seatunnel/commit/40d2c1b2132.3.6
[Feature][Connector-V2] Iceberg-sink supports writing data to branches (#6697)https://github.com/apache/seatunnel/commit/e3103535cc2.3.6
[Fix][Connector-V2] Fix connector support SPI but without no args constructor (#6551)https://github.com/apache/seatunnel/commit/5f3c9c36a52.3.5
[Improve] Add SaveMode log of process detail (#6375)https://github.com/apache/seatunnel/commit/b0d70ce2242.3.5
[Improve][Zeta] Add classloader cache mode to fix metaspace leak (#6355)https://github.com/apache/seatunnel/commit/9c3c2f183d2.3.5
[Improve][API] Unify type system api(data & type) (#5872)https://github.com/apache/seatunnel/commit/b38c7edcc92.3.5
[Feature] Supports iceberg sink #6198 (#6265)https://github.com/apache/seatunnel/commit/18d3e861942.3.5
[Test][E2E] Add thread leak check for connector (#5773)https://github.com/apache/seatunnel/commit/1f2f3fc5f02.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[BUG][Connector-V2] Iceberg source lost data with parallelism option (#5732)https://github.com/apache/seatunnel/commit/7f3b4be0752.3.4
[Dependency]Bump org.apache.avro:avro in /seatunnel-connectors-v2/connector-iceberg (#5582)https://github.com/apache/seatunnel/commit/13753a927b2.3.4
[Improve][Pom] Add junit4 to the root pom (#5611)https://github.com/apache/seatunnel/commit/7b4f7db2a22.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Doc][Iceberg] Improved iceberg documentation (#5335)https://github.com/apache/seatunnel/commit/659a68a0be2.3.4
[Hotfix] Fix com.google.common.base.Preconditions to seatunnel shade one (#5284)https://github.com/apache/seatunnel/commit/ed5eadcf732.3.3
[Hotfix][Connector][Iceberg] Fix iceberg source stream mode init error (#4638)https://github.com/apache/seatunnel/commit/64760eed4d2.3.2
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[Improve][SourceConnector] Unifie Iceberg source fields to schema (#3959)https://github.com/apache/seatunnel/commit/20e1255fab2.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Improve][Connector-V2][Iceberg] Unified exception for iceberg source connector (#3677)https://github.com/apache/seatunnel/commit/e24843515f2.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API &amp; Connector &amp; Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Feature][Connector-V2][Iceberg] Modify the scope of flink-shaded-hadoop-2 to provided to be compatible with hadoop3.x (#3046)https://github.com/apache/seatunnel/commit/b38c50789f2.3.0
[Feature][Connector V2] expose configurable options in Iceberg (#3394)https://github.com/apache/seatunnel/commit/bd9a313ded2.3.0
[Improve][Connector][Iceberg] Improve code. (#3065)https://github.com/apache/seatunnel/commit/9f38e3da742.3.0-beta
[Code-Improve][Iceberg] Use automatic resource management to replace 'try - finally' code block. (#2909)https://github.com/apache/seatunnel/commit/b7f640724b2.3.0-beta
[Feature][Connector-V2] Add iceberg source connector (#2615)https://github.com/apache/seatunnel/commit/ffc6088a792.2.0-beta