Iceberg

Sink The Iceberg connector writes streaming data to Apache Iceberg tables. It works with REST catalogs (Polaris, Nessie, Tabular, Gravitino), AWS Glue, Hive Metastore, and Google BigLake catalogs. Data lands as Parquet files on S3, GCS, or Azure storage, committed atomically through the Iceberg catalog at every checkpoint. The connector supports both append-only mode (inserts only, any parallelism) and CDC mode (inserts, updates, and deletes via equality delete files, requires parallelism: 1). Tables are auto-created on the first write if they don’t already exist - Ingest Engine infers the Iceberg schema from the pipeline’s Arrow schema, applies partitioning and table properties from your config, and registers the table in the catalog. See Shared data services for how coordinated Iceberg commits can reduce catalog pressure across pipelines.

Catalog

The catalog config tells Ingest Engine which Iceberg catalog to connect to. This is where table metadata lives - the catalog tracks schemas, partition specs, snapshots, and the location of data files.

REST Catalog

The most common choice. Works with Polaris, Nessie, Tabular, Gravitino, and any catalog that implements the Iceberg REST specification.

catalog:
  type: rest
  url: http://localhost:8181
  warehouse: s3://bucket/warehouse
  auth:
    token: "${ICEBERG_TOKEN}"

url (required) - Base URL for the REST catalog endpoint. Ingest Engine passes this as the uri property to the underlying iceberg-catalog-rest client.
warehouse (optional) - Warehouse name or location. Sent as the warehouse property to the catalog. When the table doesn’t exist yet, Ingest Engine derives the table location from this path: {warehouse}/{namespace}/{table_name}.
auth - Authentication settings. Defaults to anonymous if omitted. See below.
access_delegation (optional) - Value for the X-Iceberg-Access-Delegation header. Defaults to vended-credentials so Polaris vends per-table cloud credentials automatically. Set to empty string to suppress the header for catalogs that reject it. Set to remote-signing or a comma-separated list for other delegation modes.
extra_props (optional) - Catch-all map passed verbatim to the iceberg-catalog-rest builder. Use for properties that don’t have a first-class field yet: custom headers (header.X-Custom-Audit), future OAuth2 grant-type parameters, or catalog-specific extensions. First-class fields above take precedence over entries here; storageOptions at the sink level overrides everything.

REST Authentication

All auth fields are optional. Choose the method that matches your catalog. Anonymous (default) - no auth fields needed. Works for local development or catalogs with no authentication. Bearer token - pre-issued token that skips OAuth2:

auth:
  token: "${ICEBERG_TOKEN}"

token - Pre-issued bearer token. Supports ${ENV_VAR} substitution. When both token and credential are set, token wins (matches iceberg-catalog-rest semantics).

OAuth2 client credentials - machine-to-machine auth used by Snowflake Open Catalog, self-hosted Polaris, and external OIDC providers (Okta, Auth0, Entra, Keycloak):

auth:
  credential: "${CLIENT_ID}:${CLIENT_SECRET}"
  oauth2_server_uri: https://auth.example.com/token
  scope: PRINCIPAL_ROLE:ALL

credential - OAuth2 client_id:client_secret (colon-separated). Supports ${ENV_VAR} substitution.
oauth2_server_uri - External IdP token endpoint. If unset, the client uses the catalog’s own /v1/oauth/tokens endpoint (Polaris internal mode). Point at the IdP URL to use external or mixed mode.
scope - OAuth2 scope. Snowflake Open Catalog expects PRINCIPAL_ROLE:ALL (or a specific role). Generic IdPs use it for scope gating.
audience - OAuth2 audience. Required by Auth0 and other IdPs that scope the minted token to a specific API.
resource - OAuth2 resource. Used by Microsoft Entra / AD FS and IdPs that distinguish resource from audience.

AWS Glue Catalog

For Amazon-managed Iceberg tables. The Glue catalog maps AWS credentials to S3 automatically, so you typically don’t need storageOptions.

catalog:
  type: glue
  region: us-east-1
  warehouse: s3://my-bucket/warehouse
  access_key_id: "${AWS_ACCESS_KEY_ID}"
  secret_access_key: "${AWS_SECRET_ACCESS_KEY}"

region (required) - AWS region where the Glue catalog lives. Also used for the underlying S3 client.
warehouse (required) - S3 warehouse path. Used to derive table locations for newly created tables.
access_key_id (optional) - AWS access key ID. Supports ${ENV_VAR} substitution. If omitted, Ingest Engine uses the default AWS credential chain (environment variables, instance profile, etc.).
secret_access_key (optional) - AWS secret access key. Supports ${ENV_VAR} substitution.
session_token (optional) - AWS session token for temporary credentials. Supports ${ENV_VAR} substitution.
catalog_id (optional) - AWS Glue catalog ID. Defaults to the AWS account ID.

Hive Metastore Catalog

For Hadoop-ecosystem deployments using a Hive Metastore Thrift service.

catalog:
  type: hive
  uri: thrift://localhost:9083
  warehouse: hdfs://namenode:8020/warehouse

uri (required) - Hive Metastore Thrift URI (e.g., thrift://localhost:9083).
warehouse (required) - Warehouse location (e.g., hdfs://namenode:8020/warehouse or s3://bucket/warehouse).
principal (optional) - Kerberos principal for authenticated clusters.
keytab (optional) - Kerberos keytab file path.

Google BigLake Catalog

For GCP-managed Iceberg tables. BigLake vends GCS OAuth2 tokens automatically, so you typically don’t need storageOptions.

catalog:
  type: biglake
  project_id: my-gcp-project
  catalog_id: my-catalog
  warehouse: gs://bucket/warehouse

project_id (required) - GCP project ID.
catalog_id (required) - BigLake catalog ID.
warehouse (required) - GCS warehouse location. Must start with gs://. Ingest Engine validates this at config time.
uri (optional) - BigLake REST endpoint. Defaults to https://biglake.googleapis.com/iceberg/v1/restcatalog.
service_account (optional) - Service account name for the GCP metadata server. Defaults to default.

Sink Config

The sink config controls where data is written (namespace + table), how files are structured (rotation, parquet settings, partitioning), and whether the sink operates in append-only or CDC mode.

Table Identity

namespace: my_database
tableName: events

namespace (required) - Table namespace (database name). Supports dotted namespaces for catalogs that use them: "namespace1.namespace2". Ingest Engine auto-creates the namespace if it doesn’t exist.
tableName (required) - Iceberg table name. Combined with namespace to form the full table identifier.

Write Mode

appendOnly: true

appendOnly (boolean, default: true) - Controls how the sink processes incoming records. When true, only inserts are handled - this is faster and supports any parallelism level. When false, the sink operates in CDC mode: it handles inserts, updates, and deletes using equality delete files. CDC mode requires primary_keys defined in the schema and parallelism: 1. See CDC Mode for details.

Upload Tuning

uploadChunkSizeBytes: 8388608
uploadConcurrency: 8

uploadChunkSizeBytes (integer, default: 8388608 / 8 MB) - Multipart upload chunk size in bytes. Each file upload to object storage is split into chunks of this size. Larger chunks reduce the number of API calls; smaller chunks reduce memory usage.
uploadConcurrency (integer, default: 8) - Number of upload parts sent concurrently. Higher values increase S3/GCS upload throughput at the cost of more memory. Ingest Engine applies these values to the FileIO constructed by the catalog: file_io.with_write_chunk_size(...) and file_io.with_write_concurrency(...).

Table Creation Timeout

tableCreationTimeoutSecs: 30

tableCreationTimeoutSecs (integer, default: 30) - Timeout in seconds for non-primary workers to wait for table creation. In multi-worker append-only mode, task 0 creates the table on the first write. Other tasks poll load_table() every 500ms until the table appears or this timeout expires.

File Rotation

Controls when Parquet data files are rotated. Ingest Engine uses size-based rotation - when the estimated in-memory size of the current file exceeds the threshold, the file is closed and a new one is opened. Time-based rotation is handled implicitly by the checkpoint interval (files are flushed at every checkpoint).

fileRotation:
  maxFileSizeBytes: 134217728

maxFileSizeBytes (integer, default: 536870912 / 512 MB) - Maximum file size in bytes before rotation. Ingest Engine tracks an in-memory size estimate (batch.get_array_memory_size()) and rotates when it crosses this threshold. The actual on-disk Parquet file will be smaller than this estimate due to encoding and compression.

Parquet Writer

Fine-grained control over how Parquet files are written. These settings are applied to the WriterProperties passed to the underlying parquet-rs writer. They also auto-derive corresponding Iceberg table properties at table creation time (see Iceberg Table Properties).

parquet:
  compression: zstd
  compressionLevel: 3
  maxRowGroupRows: 500000
  dataPageSize: 1048576
  dictionaryEnabled: true
  dictionaryPageSize: 1048576

compression (string, default: uncompressed) - Parquet compression codec. Supported values: uncompressed, snappy, gzip, lz4, zstd. Maps directly to parquet::basic::Compression in the writer. Also sets the write.parquet.compression-codec Iceberg table property so other writers (Spark compaction, Trino) use the same codec.
compressionLevel (integer, optional) - Compression level for codecs that support it. Gzip: 0-9. Zstd: 1-22. Invalid values fall back to the codec default. Also sets write.parquet.compression-level on the table.
maxRowGroupRows (integer, default: 1000000) - Maximum number of rows per row group. This is a client-side setting only - Iceberg has no portable table property for row-group row count (Iceberg uses bytes, not rows), so this does not propagate to other writers.
dataPageSize (integer, default: 1048576 / 1 MB) - Target data page size in bytes. Maps to WriterProperties::set_data_page_size_limit(). Also sets write.parquet.page-size-bytes on the table.
dictionaryEnabled (boolean, default: true) - Enable dictionary encoding globally for all columns. Maps to WriterProperties::set_dictionary_enabled(). Also sets the parquet.enable.dictionary Iceberg table property.
dictionaryPageSize (integer, default: 1048576 / 1 MB) - Maximum dictionary page size in bytes. Maps to WriterProperties::set_dictionary_page_size_limit(). Also sets write.parquet.dict-size-bytes on the table.

Per-Column Dictionary Overrides

Override the global dictionaryEnabled setting for individual columns. Each entry calls WriterProperties::set_column_dictionary_enabled() for that column.

parquet:
  dictionaryEnabled: true
  dictionaryColumns:
    - name: request_id
      dictionaryEnabled: false
    - name: status_code
      dictionaryEnabled: true

dictionaryColumns[].name - Column name to override.
dictionaryColumns[].dictionaryEnabled - Whether dictionary encoding is enabled for this column.

High-cardinality columns (UUIDs, request IDs) waste memory on dictionary encoding and should be disabled. Low-cardinality columns (status codes, country codes) benefit from dictionary encoding even if the global default is off. Note: Iceberg has no per-column dictionary table property. Ingest Engine projects the overall intent into the global parquet.enable.dictionary flag - it’s set to true if the global setting is on OR any per-column override enables dictionary encoding.

Per-Column Bloom Filters

Bloom filters enable fast equality predicate pruning at the row-group level. They are configured via the bloom_filter flag on schema fields (not in the parquet config block). Ingest Engine collects all schema fields with bloom_filter: true and calls WriterProperties::set_column_bloom_filter_enabled() for each one. It also sets the corresponding write.parquet.bloom-filter-enabled.column.<name> Iceberg table property.

schema:
  fields:
    - field_name: client_ip
      field_type:
        type:
          primitive: Utf8
      bloom_filter: true
    - field_name: status_code
      field_type:
        type:
          primitive: Utf8
      bloom_filter: true

Only useful for columns that appear in point-lookup WHERE clauses (e.g., WHERE client_ip = '1.2.3.4'). Unlisted columns are unaffected.

Iceberg Table Properties

Properties synced to the Iceberg catalog at table creation time. These are stored in Iceberg’s TableMetadata and affect all writers - the Ingest Engine sink, Spark compaction jobs, and ad-hoc Spark/Trino queries. Some properties are auto-derived from the parquet config (compression, page sizes, bloom filters, dictionary encoding). The fields below control the remaining table-level behaviors.

iceberg:
  targetFileSizeBytes: 536870912
  snapshotExpireAfterHours: 72
  snapshotMinToKeep: 5
  metricsDefault: "truncate(16)"
  metricsColumns:
    user_id: full
    payload: none
  metadataCleanup: true
  metadataMaxVersions: 10
  extraProperties:
    custom.key: custom-value

targetFileSizeBytes (integer, optional) - Target file size for compaction. Maps to write.target-file-size-bytes. Used by Spark’s rewrite_data_files action to decide when files need merging. If omitted, the Iceberg default (512 MB) applies.
snapshotExpireAfterHours (integer, optional) - Expire snapshots older than this many hours. Ingest Engine converts this to milliseconds and sets history.expire.max-snapshot-age-ms.
snapshotMinToKeep (integer, optional) - Minimum snapshots to keep regardless of age. Maps to history.expire.min-snapshots-to-keep.
metricsDefault (string, optional) - Default column metrics mode for manifest entries. Valid values: truncate(16), none, counts, full. Maps to write.metadata.metrics.default.
metricsColumns (map, optional) - Per-column metrics overrides. Key is the column name, value is the mode. Maps to write.metadata.metrics.column.<name>.
metadataCleanup (boolean, optional) - Auto-delete old metadata.json files after each commit. Maps to write.metadata.delete-after-commit.enabled.
metadataMaxVersions (integer, optional) - Maximum previous metadata versions to retain. Maps to write.metadata.previous-versions-max.
extraProperties (map, optional) - Passthrough for any Iceberg table property not covered above. Set as-is on the table. Last-write-wins if keys overlap with auto-derived properties (these are applied last).

Partitioning

Iceberg partitioning controls how data files are organized on storage. Partition transforms are applied to source columns to produce partition values - queries that filter on partition columns can skip entire file groups.

partitioning:
  fields:
    - name: event_date
      transform: identity
    - name: event_time
      transform: hour
  shuffle_by_partition:
    enabled: true

fields[].name - Source column name to partition by. Must exist in the schema - Ingest Engine validates this at config time. Dots in field names are replaced with underscores for Avro compatibility (Iceberg uses Avro for manifest serialization).
fields[].transform - Partition transform to apply. Default: identity.

Supported transforms:

identity - Use the field value as-is. Best for date fields or low-cardinality string fields (region, country).
hour - Extract hour from a timestamp column. Produces partition names like event_time_hour. Good for high-volume data that needs hourly granularity.
month - Extract month from a timestamp column. Produces partition names like event_time_month. Good for monthly aggregation patterns.
year - Extract year from a timestamp column. Produces partition names like event_time_year. Good for historical data with long retention.

Partition Shuffling

shuffle_by_partition.enabled (boolean, default: false) - When enabled, Ingest Engine hash-shuffles records by partition key before they reach the sink writer. This ensures each partition’s data goes to a single writer, reducing the total number of output files. The tradeoff: shuffling adds a network exchange step and can cause backlog if data is skewed toward a few partitions.

Storage Options

Cloud storage credentials and endpoint configuration passed through to the catalog builder. These are merged into the catalog’s properties map, so the FileIO constructed by load_table() inherits them automatically. You don’t need to configure storage options when the catalog handles credentials natively (e.g., Glue maps AWS creds to S3 automatically, BigLake vends GCS tokens).

S3 / MinIO

storageOptions:
  s3.endpoint: http://minio:9000
  s3.access-key-id: "${AWS_ACCESS_KEY_ID}"
  s3.secret-access-key: "${AWS_SECRET_ACCESS_KEY}"
  s3.region: us-east-1
  s3.path-style-access: "true"

s3.endpoint - Custom S3 endpoint. Required for MinIO or S3-compatible stores.
s3.access-key-id - AWS access key ID.
s3.secret-access-key - AWS secret access key.
s3.session-token - AWS session token for temporary credentials.
s3.region - AWS region.
s3.path-style-access - Set "true" for MinIO and S3-compatible stores that require path-style URLs.
s3.disable-config-load - Skip loading config from environment variables and files.

GCS

storageOptions:
  gcs.project-id: my-gcp-project
  gcs.credentials-json: "${GCS_CREDENTIALS_BASE64}"

gcs.project-id - GCP project ID.
gcs.credentials-json - Base64-encoded service account JSON (for non-vended credential mode).
gcs.oauth2.token - OAuth2 token (manual injection, typically from vended credentials).
gcs.service.path - Custom GCS endpoint (for emulators).
gcs.allow-anonymous - Skip signing requests (for public buckets).
gcs.disable-vm-metadata - Skip loading credentials from GCE metadata server.
gcs.disable-config-load - Skip loading config from environment variables and files.

Azure

storageOptions:
  azure.account-name: mystorageaccount
  azure.account-key: "${AZURE_ACCOUNT_KEY}"

azure.account-name - Azure storage account name.
azure.account-key - Azure storage account key.
azure.sas-token - Azure SAS token.
azure.tenant-id - Azure AD tenant ID.
azure.client-id - Azure AD client ID.
azure.client-secret - Azure AD client secret.

CDC Mode

To handle updates and deletes (not just inserts), set appendOnly: false and define primary_keys in the schema. CDC mode requires parallelism: 1 because all writes must go through a single writer to maintain consistency. In CDC mode, Ingest Engine writes equality delete files using the primary key columns. The Iceberg spec prohibits float and double types as identifier fields (due to NaN and precision issues) - Ingest Engine validates this at config time and rejects float/double primary keys.

namespace: cdc_db
tableName: payments
appendOnly: false

With a schema that defines primary keys:

schema:
  format:
    parquet: {}
  fields:
    - field_name: id
      field_type:
        type:
          primitive: Int64
      nullable: false
    - field_name: amount
      field_type:
        type:
          primitive: F64
      nullable: false
  primary_keys:
    - id

Complete Example

A full configuration writing JSON events to a partitioned Iceberg table on S3 via a REST catalog, with Zstd compression and bloom filters on the client_ip column.

catalog:
  type: rest
  url: http://polaris:8181
  warehouse: s3://data-lake/warehouse
  auth:
    credential: "${POLARIS_CLIENT_ID}:${POLARIS_CLIENT_SECRET}"
    scope: PRINCIPAL_ROLE:ALL
namespace: analytics
tableName: web_events
appendOnly: true
storageOptions:
  s3.region: us-east-1
  s3.access-key-id: "${AWS_ACCESS_KEY_ID}"
  s3.secret-access-key: "${AWS_SECRET_ACCESS_KEY}"
fileRotation:
  maxFileSizeBytes: 134217728
parquet:
  compression: zstd
  compressionLevel: 3
  maxRowGroupRows: 500000
  dictionaryEnabled: true
  dictionaryColumns:
    - name: request_id
      dictionaryEnabled: false
partitioning:
  fields:
    - name: event_date
      transform: identity
    - name: region
      transform: identity
  shuffle_by_partition:
    enabled: true
iceberg:
  targetFileSizeBytes: 268435456
  snapshotExpireAfterHours: 168
  snapshotMinToKeep: 3
  metricsDefault: "truncate(16)"
  metadataCleanup: true

With schema:

schema:
  format:
    parquet: {}
  fields:
    - field_name: event_id
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: client_ip
      field_type:
        type:
          primitive: Utf8
      nullable: true
      bloom_filter: true
    - field_name: event_date
      field_type:
        type:
          primitive: Date32
      nullable: false
    - field_name: region
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: payload
      field_type:
        type:
          primitive: Utf8
      nullable: true

See Schema Reference for field type details.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Catalog

REST Catalog

REST Authentication

AWS Glue Catalog

Hive Metastore Catalog

Google BigLake Catalog

Sink Config

Table Identity

Write Mode

Upload Tuning

Table Creation Timeout

File Rotation

Parquet Writer

Per-Column Dictionary Overrides

Per-Column Bloom Filters

Iceberg Table Properties

Partitioning

Partition Shuffling

Storage Options

S3 / MinIO

GCS

Azure

CDC Mode

Complete Example

JSON Schema Reference

​Catalog

​REST Catalog

​REST Authentication

​AWS Glue Catalog

​Hive Metastore Catalog

​Google BigLake Catalog

​Sink Config

​Table Identity

​Write Mode

​Upload Tuning

​Table Creation Timeout

​File Rotation

​Parquet Writer

​Per-Column Dictionary Overrides

​Per-Column Bloom Filters

​Iceberg Table Properties

​Partitioning

​Partition Shuffling

​Storage Options

​S3 / MinIO

​GCS

​Azure

​CDC Mode

​Complete Example

​JSON Schema Reference

Catalog

REST Catalog

REST Authentication

AWS Glue Catalog

Hive Metastore Catalog

Google BigLake Catalog

Sink Config

Table Identity

Write Mode

Upload Tuning

Table Creation Timeout

File Rotation

Parquet Writer

Per-Column Dictionary Overrides

Per-Column Bloom Filters

Iceberg Table Properties

Partitioning

Partition Shuffling

Storage Options

S3 / MinIO

GCS

Azure

CDC Mode

Complete Example

JSON Schema Reference