Cloud
Also in Self-Managed

iceberg

Writes data to Apache Iceberg tables using the REST catalog API.

This output supports:

  • Multiple storage backends (S3, GCS, Azure)

  • Automatic table creation with schema detection

  • Partition transforms (year, month, day, hour, bucket, truncate)

  • Schema evolution (automatic column addition)

  • Transaction retry logic for concurrent writes

This output is designed to work with REST catalog implementations like Apache Polaris, AWS Glue Data Catalog, and the Databricks Unity Catalog.

Tables are written using version 2 of the Iceberg specification. A pre-existing version 1 table is upgraded to version 2 automatically on first write, irreversibly, unless merge_strategy is copy-on-write, which only ever writes plain data files and therefore works on version 1 or version 2 tables without forcing the upgrade (see Merge strategies).

Catalog integration

Apache Polaris

To use with Apache Polaris:

  • Set catalog.url to the Polaris REST endpoint (for example, http://localhost:8181/api/catalog).

  • Set catalog.warehouse to the catalog name configured in Polaris.

  • Configure catalog.auth.oauth2 with client credentials granted access to the catalog.

AWS Glue Data Catalog

To use with AWS Glue Data Catalog:

  • Set catalog.url to https://glue.<region>.amazonaws.com/iceberg (the REST client appends the API version automatically).

  • Set catalog.warehouse to your AWS account ID (the Glue catalog identifier).

  • Set schema_evolution.table_location to an S3 prefix (for example, s3://my-bucket/; a trailing slash is optional) since Glue does not automatically assign table locations.

  • Configure catalog.auth.aws_sigv4 with the appropriate region and set service to glue.

  • Configure storage.aws_s3 with the same bucket and region.

Storage

Azure Blob Storage (ADLS Gen2)

To use with Azure Data Lake Storage Gen2:

  • Configure storage.azure_blob_storage with your storage account name and container.

  • Authenticate using one of: storage_access_key (shared key), storage_sas_token, or storage_connection_string.

  • The storage account must have hierarchical namespace (HNS) enabled for ADLS Gen2 compatibility.

Type mapping

Values are mapped from Bloblang types to Iceberg types as follows:

Bloblang type Iceberg type

string

string

bytes

binary

bool

boolean

number

double

timestamp

timestamp (with timezone)

object

struct

array

list

Row-level operations

By default this output is append-only: every message becomes a new row (row_operation: insert), and existing configurations are unaffected. Set row_operation to apply a per-message operation, with identifier_fields defining the row identity:

  • insert: append the row. This is an unconditional append: it is not keyed or de-duplicated.

  • upsert: replace any existing rows matching identifier_fields, then append this row (equivalent to Iceberg’s Flink UPSERT mode).

  • delete: remove rows matching identifier_fields.

row_operation supports interpolation, so the operation can be driven by the data itself (for example by mapping a change-data-capture stream’s operation field), but no CDC-specific format is assumed (see the change-data-capture example below). It is named row_operation to distinguish it from Iceberg’s snapshot-level operation.

upsert and delete require identifier_fields. How those mutations are materialised on disk, and, critically, which query engines can then read the table, is controlled by merge_strategy. See Merge strategies: merge-on-read vs copy-on-write below for the decision guide, support matrix, and maintenance guidance.

Identifier fields. identifier_fields must reference existing table columns of a primitive, non-floating-point type. A static upsert/delete is validated at startup; an interpolated row_operation is validated per message at write time, so an empty identifier_fields is not caught until the first upsert/delete message arrives. Identifier columns of a temporal type (timestamp, timestamptz, date, time) are interpreted exactly like their data columns: time values and RFC 3339 strings are unambiguous, and a bare numeric epoch is interpreted using the batch’s schema metadata when present, falling back to the same default units the insert path applies (the key is interpreted by the identical rules as the insert path, so it encodes what the insert stored whenever both batches carry the same unit declaration). Beware feeding numeric epochs without schema metadata in a unit other than that default: the key would silently match nothing; set require_schema_metadata: true to make an undeclared unit fail loudly instead, or convert to a timestamp upstream. If the table is partitioned, merge-on-read additionally requires every partition source column to be one of the identifier_fields, since equality deletes are partition-scoped; copy-on-write carries no such restriction (it rewrites whole files by filter and routes new rows by value, and can even move a key across partitions). See the support matrix for the merge-key types each strategy accepts.

Strict key and value typing. Merge keys and row values on the upsert/delete paths are validated as strictly as the insert path has always been: a string-typed integer or boolean (for example {"id": "42"} into a long column, or "true" into a boolean column) previously worked by accident on the mutation paths and now fails the batch with an actionable error, matching the insert path’s rejections. Convert such values upstream (for example with Bloblang’s .number() or .bool()) if a source emits stringified scalars.

Under merge-on-read, when this output auto-creates a table (via schema_evolution), the identifier_fields columns are created as required and registered as the table’s Iceberg identifier-field-ids, so downstream engines and other writers see the primary key. A consequence is that a null or missing value in an identifier column is rejected on write, even for insert; identifier columns must therefore be present at creation: in the first message or declared via schema_metadata. Under copy-on-write the identifier fields are used only as the connector-side merge key and are not registered as identifier-field-ids, so auto-created columns are not forced required (this is also what lets engine-backed catalogs such as the Databricks Unity Catalog, which rejects identifier-field-ids at table creation, accept the CREATE TABLE). Pre-existing tables are never modified.

Batching and ordering. Within a single batch the last upsert/delete per identifier_fields key wins. Each batch containing an upsert/delete is committed as its own snapshot (these commits are never coalesced, which is required for correctness), so a high-throughput mutation workload produces one snapshot per batch. Size batches accordingly and run regular table maintenance (snapshot expiry and compaction) to keep metadata manageable. Pure insert-only batches keep the original append fast path, which does coalesce commits.

Ordering only holds within a batch. With more than one batch in flight, concurrent batches can commit out of order, so a stale upsert may overwrite a newer one for the same key. Set max_in_flight: 1 for keyed (change-data-capture) workloads to preserve per-key order: this is enforced by config linting whenever row_operation is anything other than a static insert.

insert is an unconditional append and is not keyed or de-duplicated. For keyed data (including change-data-capture), map create/read events to upsert, never insert: mixing insert with upsert/delete on the same key in one batch produces duplicate rows.

Merge strategies: merge-on-read vs copy-on-write

merge_strategy controls how upsert/delete mutations are written, which sets both the write cost and, most importantly, which query engines can read the table.

  • merge-on-read (the default) writes Iceberg v2 equality-delete files and applies them at read time. Writes stay cheap and streaming-friendly, but only catalog-native / Flink-world engines (Apache Polaris, Flink, Trino, Spark) can read equality deletes. Engine-backed catalogs (Snowflake and the Databricks Unity Catalog) cannot. It requires table format version 2, so a version-1 table is automatically upgraded to version 2 on the first upsert/delete; this upgrade is irreversible.

  • copy-on-write instead rewrites whole data files so the table only ever holds plain data files: no equality- or positional-delete files. Every engine that reads Iceberg data files can read the result: Snowflake and the Databricks Unity Catalog as well as Polaris, Flink and Trino. Because it writes only plain data files it works on version-1 or version-2 tables and never forces the irreversible v1→v2 upgrade.

Which to choose.

  • Choose merge-on-read for streaming or high-throughput mutation into a lake read by Polaris, Flink, Trino or Spark, where write cost must stay low.

  • Choose copy-on-write when the table must be correct on Snowflake or the Databricks Unity Catalog (or any engine that cannot read equality deletes), and the workload is batch or moderate-throughput so the write amplification is acceptable.

Copy-on-write support matrix.

  • Column types: all flat primitives (boolean, int, long, float, double, string, date, time, timestamp, timestamptz, decimal, uuid, binary, fixed), and nested struct/list/map columns whose leaves are all supported primitives.

  • Merge-key (identifier_fields) types: int, long, string, date, time, timestamp, timestamptz and uuid. decimal and boolean merge keys are not supported and error with a message pointing you at merge-on-read (an upstream limitation in the Iceberg library’s overwrite filter; it cannot apply a decimal or boolean predicate when rewriting files); both are fine as non-key columns.

  • Partitioned tables: supported, with no requirement that the partition columns be a subset of identifier_fields. A copy-on-write upsert can even move a key from one partition to another.

  • Table format: version 1 or version 2, with no forced upgrade.

Write amplification and throughput. copy-on-write rewrites every data file that contains a row for a touched key, so the cost of a batch depends on how many files its keys' rows are spread across, not on how many keys it touches: keys that cluster into a few files rewrite only those files, and even one key can rewrite many files if its rows (for example, duplicates from insert) are spread across them. Touching even a single key in a file rewrites that whole file, so a one-row change to a 512 MB file rewrites all 512 MB. To keep amplification low, sort the table by the identifier key so a batch’s keys cluster into as few files as possible, and use large batches. This is a batch / moderate-throughput mode, not a streaming one.

Memory. Under copy-on-write the whole new-row batch is materialised in memory as a single Arrow record while the batch commits, so a keyed batch’s memory scales with its total row bytes. Size keyed batches to stay within the process memory budget rather than making them arbitrarily large.

Maintenance. Because every mutating batch rewrites files and adds a snapshot, a high-churn copy-on-write workload accumulates data files and snapshots quickly. Run regular table maintenance: compaction (rewrite / bin-pack data files), snapshot expiry, and orphan-file removal. Orphan-file removal matters specifically for commits that fail ambiguously (the catalog may or may not have recorded them): the connector deliberately never deletes those attempts' newly-written data files, because the commit may still land and reference them; they are left for orphan-file removal to reclaim. The connector’s own best-effort cleanup covers only commits whose failure was a definitive catalog rejection (and superseded attempts of a retried success). Setting commit.cleanup_on_failure to false turns that connector-side cleanup off altogether, on every write path, which makes periodic orphan-file removal mandatory rather than merely advisable.

Commit retries. Leave the standard Iceberg table property commit.retry.num-retries at its default 0 on tables this output writes to. The output’s own commit retry (commit.max_retries) re-stages every attempt from a freshly reloaded table snapshot, which is safe under concurrent writers; the library-level replay a non-zero commit.retry.num-retries arms instead re-submits the original updates built against a now-stale snapshot, and under concurrent writers that replay can drop a peer commit’s files from the table state. The output logs a warning when it finds the property armed on a table it writes to.

Copy-on-write limitations.

  • A decimal merge key is not supported: use merge-on-read for a decimal key (a decimal non-key column is fine).

  • Schema evolution covers new top-level columns only; new fields appearing inside an existing nested struct/list/map column are not auto-surfaced for evolution.

  • It is a batch / moderate-throughput mode: expect heavy write amplification under scattered, high-frequency keyed mutations.

  • Tables pinned to the legacy timestamp encoding whose schema contains a no-timezone timestamp column reject upsert/delete: see Timestamp encoding on existing tables for why and for the migration path.

  • A table whose write.delete.mode property is explicitly set to merge-on-read (for example by a Spark TBLPROPERTIES clause) rejects copy-on-write mutations with an actionable error: the connector will not silently override a table-level setting other engines read, and honouring it would produce the delete files this strategy exists to avoid. Unset the property, set it to copy-on-write, or use merge_strategy: merge-on-read. (Previous releases silently overrode the property.)

Timestamp encoding on existing tables

Older versions of this output annotated no-timezone timestamp columns in the parquet files they wrote with isAdjustedToUTC=true: the annotation the Iceberg spec reserves for timestamptz. The stored microsecond instants are correct, and appends and most readers are unaffected, but the annotation makes some readers treat the column as UTC-adjusted, and it prevents copy-on-write from rewriting those files (the file’s annotation reads back as timestamptz, which cannot be written into a timestamp column). Current versions write the spec-correct isAdjustedToUTC=false.

To guarantee an existing table never ends up with a mix of the two annotations, the encoding is pinned per table via the table property redpanda-connect.timestamp-encoding (spec or legacy):

  • Tables created by this output carry redpanda-connect.timestamp-encoding: spec from creation.

  • For an existing table without the property, the output resolves the encoding automatically on first contact and stamps the result onto the table: if the schema has no no-timezone timestamp column, or the table has no data files, it resolves spec; otherwise the output inspects one data file’s parquet footer and adopts whatever that file already contains (legacy for isAdjustedToUTC=true). A table that cannot be probed fails the write rather than risk mixing annotations, whether because a footer is unreadable, or because the first 1000 parquet data files probed (in manifest order) carry no no-timezone timestamp column at all (possible when the column was added by later schema evolution on a large table); in the latter case the error names the fix: set the property on the table explicitly and the probe never runs. The stamp itself is a SetProperties catalog commit, so the principal this output authenticates as needs permission to write table properties in addition to writing data; a principal without it turns a previously-healthy insert pipeline into a hard failure on first contact after upgrading, until the permission is granted or the property is set out-of-band. On a catalog that forbids all custom table properties, the set-it-explicitly escape hatch is unavailable by the same rule (the redpanda-connect. prefix carries connector semantics and is never silently stripped), so such writes fail loudly rather than risk mixing annotations.

  • Once stamped, the property is authoritative and the probe never runs again. An unrecognised property value is a hard error.

A table pinned legacy keeps receiving the legacy annotation on every new file, byte-identical to what previous releases wrote, so appends and merge-on-read continue working unchanged forever. The one restriction is mutating copy-on-write (upsert/delete): it must rewrite existing files, which the legacy annotation prevents, so such writes fail upfront with an actionable error (pure insert batches still work). To migrate a legacy table to the spec encoding: rewrite/compact the table’s data files with an engine that writes the spec annotation (for example, Spark’s rewrite_data_files), then set the table property redpanda-connect.timestamp-encoding to spec, keeping any running instances of this output that write to the table stopped (or restarting them) around the migration: a live writer only re-reads the property when its writer is recreated. A table whose existing files already mix both annotations (for example one written to by several engines or connector versions over time) can be pinned either way by the probe, depending on which file it happens to read first, and a copy-on-write mutation on such a table may then fail mid-rewrite with the underlying library’s type-promotion error rather than the upfront migration message: compact or rewrite such a table to a single encoding before mutating it. Alternatively, keep the table on merge-on-read.

Data file compression

Compression of the parquet data files this output writes is resolved per table, in this order:

  1. parquet.compression, if you set it.

  2. otherwise the table’s own write.parquet.compression-codec property, if the table has one.

  3. otherwise uncompressed.

parquet.compression is optional rather than defaulted precisely so that step 2 is reachable: an unset field means "whatever the table says", which is not the same as explicitly choosing uncompressed.

Prefer the table property for compression settings whenever possible. parquet.compression only applies to files this connector writes directly. The Iceberg library can perform additional writes, such as copy-on-write rewrites and equality-delete files, and these do not honour this field. They follow the table’s write.parquet.compression-codec property instead, which defaults to zstd. Setting the table property is the only way to get one codec for every file in the table. Use this field when you cannot set table properties, for example on Databricks Unity Catalog.

The uncompressed default applies only to a table whose property is genuinely absent. A table created through the Iceberg library (which includes tables this output creates itself) comes back carrying write.parquet.compression-codec: zstd, set at creation, so an otherwise unconfigured pipeline writing to such a table produces zstd throughout rather than anything uncompressed. Either way a table may end up holding a mixture of codecs, which is legal and transparent: parquet records its codec per column chunk, readers handle mixed files, and changing compression never requires rewriting existing data.

Codec support. snappy, gzip and zstd are read by every engine this output targets. Parquet permits others that are not offered here: the original lz4 codec was ambiguously specified and readers disagree on what it means, lz4_raw is its unambiguous replacement but has younger and less universal reader support, and brotli and lzo are patchily supported. If the table property names one of those it is reported in the log and the files this output writes are uncompressed instead. That governs only those files; the Iceberg library does map those codecs, so if the intent is that nothing writes one, change the table property rather than relying on this.

Cost. Negligible impact on write throughput; the size savings depend heavily on how compressible your data is.

Performance

This output benefits from sending multiple messages in flight in parallel for improved performance. You can tune the max number of in flight messages (or message batches) with the field max_in_flight.

This output benefits from sending messages as a batch for improved performance. Batches can be formed at both the input and output level. You can find out more in this doc.

  • Common

  • Advanced

output:
  label: ""
  iceberg:
    catalog:
      url: "" # No default (required)
      warehouse: "" # No default (optional)
      auth:
        oauth2:
          server_uri: /v1/oauth/tokens
          client_id: "" # No default (required)
          client_secret: "" # No default (required)
          scope: "" # No default (optional)
        bearer: "" # No default (optional)
        aws_sigv4: {}
    namespace: "" # No default (required)
    table: "" # No default (required)
    storage:
      aws_s3:
        bucket: "" # No default (required)
        region: "" # No default (optional)
        endpoint: "" # No default (optional)
      gcp_cloud_storage:
        bucket: "" # No default (required)
        credentials_type: "" # No default (optional)
        credentials_file: "" # No default (optional)
        credentials_json: "" # No default (optional)
      azure_blob_storage:
        storage_account: "" # No default (required)
        container: "" # No default (required)
        storage_sas_token: "" # No default (optional)
        storage_connection_string: "" # No default (optional)
        storage_access_key: "" # No default (optional)
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
    max_in_flight: 4
output:
  label: ""
  iceberg:
    catalog:
      url: "" # No default (required)
      warehouse: "" # No default (optional)
      auth:
        oauth2:
          server_uri: /v1/oauth/tokens
          client_id: "" # No default (required)
          client_secret: "" # No default (required)
          scope: "" # No default (optional)
        bearer: "" # No default (optional)
        aws_sigv4:
          region: "" # No default (optional)
          endpoint: "" # No default (optional)
          tcp:
            connect_timeout: 0s
            keep_alive:
              idle: 15s
              interval: 15s
              count: 9
            tcp_user_timeout: 0s
          credentials:
            profile: "" # No default (optional)
            id: "" # No default (optional)
            secret: "" # No default (optional)
            token: "" # No default (optional)
            from_ec2_role: false # No default (optional)
            role: "" # No default (optional)
            role_external_id: "" # No default (optional)
          service: "" # No default (optional)
      headers: {} # No default (optional)
      tls_skip_verify: false
    namespace: "" # No default (required)
    table: "" # No default (required)
    case_sensitive_columns: true
    row_operation: insert
    identifier_fields: []
    merge_strategy: merge-on-read
    storage:
      aws_s3:
        bucket: "" # No default (required)
        region: "" # No default (optional)
        endpoint: "" # No default (optional)
        force_path_style_urls: false
        credentials:
          id: "" # No default (optional)
          secret: "" # No default (optional)
          token: "" # No default (optional)
      gcp_cloud_storage:
        bucket: "" # No default (required)
        endpoint: "" # No default (optional)
        credentials_type: "" # No default (optional)
        credentials_file: "" # No default (optional)
        credentials_json: "" # No default (optional)
      azure_blob_storage:
        storage_account: "" # No default (required)
        container: "" # No default (required)
        endpoint: "" # No default (optional)
        storage_sas_token: "" # No default (optional)
        storage_connection_string: "" # No default (optional)
        storage_access_key: "" # No default (optional)
    schema_evolution:
      enabled: false
      partition_spec: ()
      table_location: "" # No default (optional)
      schema_metadata: ""
      new_column_type_mapping: "" # No default (optional)
      require_schema_metadata: false
    commit:
      manifest_merge_enabled: true
      max_snapshot_age: 24h
      max_retries: 3
      cleanup_on_failure: true
    parquet:
      string_encoding: delta_length_byte_array
      compression: "" # No default (optional)
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)
    max_in_flight: 4

Fields

batching

Configure a batching policy.

Type: object

# Examples:
batching:
  byte_size: 5000
  count: 0
  period: 1s

# ---

batching:
  count: 10
  period: 1s

# ---

batching:
  check: this.contains("END BATCH")
  count: 0
  period: 1m

batching.byte_size

The maximum total size (in bytes) that a batch can reach before it is flushed. When the combined size of all messages in the batch reaches or exceeds this limit, the batch is immediately sent to the next stage (such as a processor or output).

Set to 0 to disable size-based batching. When disabled, messages are flushed based on other conditions (such as count or period).

Type: int

Default: 0

batching.check

A Bloblang query that returns a boolean value indicating whether a message should end a batch.

Type: string

Default: ""

# Examples:
check: this.type == "end_of_transaction"

batching.count

The number of messages at which the batch is flushed. Set to 0 to disable count-based batching.

Type: int

Default: 0

batching.period

The length of time after which an incomplete batch is flushed regardless of its size. This field accepts Go duration format strings such as 100ms, 1s, or 5s. Supported time units are ns, us, ms, s, m, and h.

Type: string

Default: ""

# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms

batching.processors[]

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. All resulting messages are flushed as a single batch, so splitting the batch into smaller batches with these processors has no effect.

Type: array<processor>

# Examples:
processors:
  - archive:
      format: concatenate


# ---

processors:
  - archive:
      format: lines


# ---

processors:
  - archive:
      format: json_array

case_sensitive_columns

Controls how message field names are matched against table column names, and how column references in the partition spec are resolved. When true (the default), names must match exactly. When false, matching is case-insensitive; set this when your downstream catalog or query engine treats column names as case-insensitive (the iceberg specification’s recommended convention) so that, for example, a message keyed "COLUMN" lands in an existing column rather than triggering schema evolution. Ambiguous case-only duplicates in the input are rejected.

Type: bool

Default: true

catalog

REST catalog configuration.

Type: object

catalog.auth

Authentication configuration for the REST catalog. Only one authentication method can be active at a time.

Type: object

catalog.auth.aws_sigv4

AWS SigV4 authentication (for AWS Glue Data Catalog or API Gateway).

Type: object

catalog.auth.aws_sigv4.credentials

Manually configure the AWS credentials to use (optional). For more information, see the Amazon Web Services guide.

Type: object

catalog.auth.aws_sigv4.credentials.from_ec2_role

Use the credentials of a host EC2 machine configured to assume an IAM role associated with the instance.

Type: bool

catalog.auth.aws_sigv4.credentials.id

The ID of the AWS credentials to use.

Type: string

catalog.auth.aws_sigv4.credentials.profile

The profile from ~/.aws/credentials to use.

Type: string

catalog.auth.aws_sigv4.credentials.role

The ARN of the role to assume.

Type: string

catalog.auth.aws_sigv4.credentials.role_external_id

An external ID to use when assuming a role.

Type: string

catalog.auth.aws_sigv4.credentials.secret

The secret for the AWS credentials in use.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

catalog.auth.aws_sigv4.credentials.token

The token for the AWS credentials in use. Required only when using short-term credentials.

Type: string

catalog.auth.aws_sigv4.endpoint

A custom endpoint URL for AWS API requests. Use this to connect to AWS-compatible services or local testing environments instead of the standard AWS endpoints.

Type: string

catalog.auth.aws_sigv4.region

The AWS region in which your resources are hosted.

Type: string

catalog.auth.aws_sigv4.service

AWS service name for SigV4 signing.

Type: string

catalog.auth.aws_sigv4.tcp

Configure TCP socket-level settings to optimize network performance and reliability. These low-level controls are useful for:

  • Unresponsive hosts: Set connect_timeout to limit how long a connection attempt can take (the default 0s sets no limit)

  • Long-lived connections: Configure keep_alive settings to detect and recover from stale connections

  • Unstable networks: Tune keep-alive probes to balance between quick failure detection and avoiding false positives

  • Linux systems with specific requirements: Use tcp_user_timeout (Linux 2.6.37+) to control data acknowledgment timeouts

Most users should keep the default values. Only modify these settings if you’re experiencing connection stability issues or have specific network requirements.

Type: object

catalog.auth.aws_sigv4.tcp.connect_timeout

Maximum amount of time a dial will wait for a connect to complete. Zero disables.

Type: string

Default: 0s

catalog.auth.aws_sigv4.tcp.keep_alive

TCP keep-alive probe configuration.

Type: object

catalog.auth.aws_sigv4.tcp.keep_alive.count

Maximum unanswered keep-alive probes before dropping the connection. Zero defaults to 9.

Type: int

Default: 9

catalog.auth.aws_sigv4.tcp.keep_alive.idle

Duration the connection must be idle before sending the first keep-alive probe. Zero defaults to 15s. Negative values disable keep-alive probes.

Type: string

Default: 15s

catalog.auth.aws_sigv4.tcp.keep_alive.interval

Duration between keep-alive probes. Zero defaults to 15s.

Type: string

Default: 15s

catalog.auth.aws_sigv4.tcp.tcp_user_timeout

Maximum time to wait for acknowledgment of transmitted data before killing the connection. Linux-only (kernel 2.6.37+), ignored on other platforms. When enabled, keep_alive.idle must be greater than this value per RFC 5482. Zero disables.

Type: string

Default: 0s

catalog.auth.bearer

Static bearer token for authentication. For testing only, not recommended for production.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

catalog.auth.oauth2

OAuth2 authentication configuration.

Type: object

catalog.auth.oauth2.client_id

OAuth2 client identifier.

Type: string

catalog.auth.oauth2.client_secret

OAuth2 client secret.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

catalog.auth.oauth2.scope

OAuth2 scope to request.

Type: string

catalog.auth.oauth2.server_uri

OAuth2 token endpoint URI.

Type: string

Default: /v1/oauth/tokens

catalog.headers

Custom HTTP headers to include in all requests to the catalog.

Type: object<string>

# Examples:
headers:
  X-Api-Key: your-api-key

catalog.tls_skip_verify

Skip TLS certificate verification. Not recommended for production.

Type: bool

Default: false

catalog.url

The REST catalog endpoint URL.

Type: string

# Examples:
url: http://localhost:8181/api/catalog

# ---

url: https://polaris.example.com/api/catalog

# ---

url: https://glue.us-east-1.amazonaws.com/iceberg

catalog.warehouse

The REST catalog warehouse.

Type: string

# Examples:
warehouse: redpanda-catalog

commit

Commit behavior configuration.

Type: object

commit.cleanup_on_failure

Whether to remove the files a failed commit had already written. A commit’s data files (and, under merge-on-read, its equality-delete files) are written to storage before the catalog commit, so a failed commit leaves them referenced by no snapshot; the default true removes them best-effort to limit orphaned objects. The same sweep reclaims the superseded files of earlier attempts when a copy-on-write commit succeeds on retry (each attempt re-stages fresh rewrites). This applies to every write path: append, merge-on-read, and copy-on-write.

Cleanup is already skipped automatically whenever a commit’s outcome is ambiguous, since such a commit may still land server-side, and deleting files a landed snapshot references would corrupt the table. Those leftovers are always deferred to table maintenance regardless of this setting.

Setting this to false disables cleanup entirely, as a safety valve: writes then never delete anything, and every failed commit leaves its files orphaned in storage for Iceberg orphan-file maintenance (snapshot expiry plus remove_orphan_files) to reclaim. See the Merge strategies section above for that maintenance guidance.

Type: bool

Default: true

commit.manifest_merge_enabled

Merge small manifest files during commits to reduce metadata overhead.

Type: bool

Default: true

commit.max_retries

Maximum number of times to retry a failed transaction commit.

Type: int

Default: 3

commit.max_snapshot_age

Maximum age of snapshots to retain for time-travel queries. Set to zero to disable removing old snapshots.

Type: string

Default: 24h

identifier_fields[]

The columns forming the row identity (the Iceberg identifier fields, or equality-delete key) used by upsert and delete. Required when row_operation can evaluate to upsert or delete, and must reference existing table columns of a primitive, non-floating-point type.

See the Row-level operations section above for the full constraints, including the temporal-type and partitioning rules and when the requirement is enforced.

Type: array<string>

Default: []

# Examples:
identifier_fields:
  - id

# ---

identifier_fields:
  - tenant_id
  - user_id

max_in_flight

The maximum number of messages to have in flight at a given time. For outputs that send messages in batches, this limit applies to message batches. Increase this value to improve throughput.

Type: int

Default: 4

merge_strategy

How upsert and delete are materialised on disk.

  • merge-on-read (the default) writes Iceberg v2 equality-delete files. Deletes are applied at read time, so writes stay cheap and streaming-friendly, but only catalog-native / Flink-world engines can read the result. Engine-backed catalogs such as Snowflake and the Databricks Unity Catalog cannot read equality deletes.

  • copy-on-write rewrites whole data files so the table only ever contains plain data files (no delete files), which every engine can read, including Snowflake and Databricks Unity Catalog. It works on version-1 or version-2 tables and never forces the irreversible v1→v2 upgrade. The trade-off is heavy write amplification: each mutating batch rewrites every data file that contains a touched key, so it is a batch / moderate-throughput mode. Sort the table by the identifier key and use large batches so each rewrite touches as few files as possible.

See the Merge strategies section above for the full decision guide, copy-on-write support matrix (column and merge-key types, partitioning, table format), and maintenance guidance.

Type: string

Default: merge-on-read

Options: merge-on-read, copy-on-write

namespace

The Iceberg namespace for the table, dot delimiters are split as nested namespaces.

This field supports interpolation functions.

Type: string

# Examples:
namespace: analytics.events

# ---

namespace: production

parquet

Parquet writer configuration.

Type: object

parquet.compression

The compression codec for data files this output writes. Optional on purpose: when it is not set, the codec is taken from the table’s own write.parquet.compression-codec property, and when that is absent too, data files are written uncompressed.

Setting the table property rather than this field is usually the better choice, because the property is also honoured by the copy-on-write rewrite path (which writes its files inside the Iceberg library, out of reach of this field). The property is what gets every file in the table onto one codec, whereas this field governs only the data files this output writes itself: appends and merge-on-read data files, not copy-on-write rewrites or equality-delete files. Use this field when the property cannot be set, for example on catalogs that reject client-set table properties.

If you rely on the table property instead, use a lower-case codec name: this field accepts any casing, but the Iceberg library’s own property lookup is lower-case only and silently falls back to uncompressed for anything else. A property of ZSTD would give you compressed appends and uncompressed copy-on-write rewrites.

Only codecs that every engine this output targets can read are offered. If the table property names something else (lz4, lz4_raw, brotli, lzo), it is reported in the log and the data files this output writes are uncompressed instead.

Measurement found no throughput penalty worth planning around, at one core or four; the size benefit though is entirely data-dependent, ranging from ~15x smaller on a repetitive record shape to ~2% on random content. Note also that the uncompressed fallback applies only when the table property is absent, and tables created through the Iceberg library carry a zstd property by default. See Data file compression for the resolution order and the copy-on-write caveat.

Type: string

Options: uncompressed, snappy, gzip, zstd

parquet.string_encoding

The encoding to use for string and binary columns. Use plain for compatibility with readers that do not support DELTA_LENGTH_BYTE_ARRAY encoding, such as AWS Redshift Spectrum.

Type: string

Default: delta_length_byte_array

Options: plain, delta_length_byte_array

row_operation

The row-level operation to apply for each message: insert (append), upsert (replace rows matching identifier_fields, then append), or delete (remove rows matching identifier_fields). Supports interpolation so the operation can be driven by the data, such as a change-data-capture stream’s operation field. Defaults to insert, preserving the original append-only behavior.

See the Row-level operations section above for the full semantics, the format-version-2 upgrade, batching behavior, and important caveats.

This field supports interpolation functions.

Type: string

Default: insert

# Examples:
row_operation: insert

# ---

row_operation: ${! metadata("op") }

# ---

row_operation: '${! this.op == "d" ? "delete" : "upsert" }'

schema_evolution

Schema evolution configuration.

Type: object

schema_evolution.enabled

Enable automatic schema evolution. When enabled, new columns will be automatically added to the table.

Type: bool

Default: false

schema_evolution.new_column_type_mapping

An optional Bloblang mapping to customize column types during schema evolution. This mapping is executed for each new column and can override the inferred or schema-metadata-derived type. The mapping receives an object with fields name (column name), path (dot-separated path), value (sample value), inferred_type (the type that would be used without this mapping), message (the full message body), namespace, and table. It must return a string with a valid Iceberg type name: boolean, int, long, float, double, string, binary, date, time, timestamp, timestamptz, uuid, decimal(p,s), or fixed[n].

Type: string

schema_evolution.partition_spec

A Bloblang expression to evaluate when a new table is created to determine the table’s partition spec. The result of the mapping should be an Iceberg partition spec in the same string format as the Redpanda Streaming topic property.

This field supports interpolation functions.

Type: string

Default: ()

# Examples:
partition_spec: (col1)

# ---

partition_spec: (nested.col)

# ---

partition_spec: (year(my_ts_col))

# ---

partition_spec: (year(my_ts_col), col2)

# ---

partition_spec: (hour(my_ts_col), truncate(42, col2))

# ---

partition_spec: (day(my_ts_col), bucket(4, nested.col))

# ---

partition_spec: (day(my_ts_col), void(`non.nested column.with.dots`), identity(nested.column))

schema_evolution.require_schema_metadata

When true, writing a numeric value into a timestamp, timestamptz, date, or time column without schema_metadata registered for that column is a hard error. The default false permits a fallback path that interprets bare numeric timestamps as Unix seconds and bare numeric times as already-microseconds; convenient, but silently wrong if upstream produced milliseconds. Enable this when you cannot guarantee the upstream attaches schema metadata and want to fail loudly rather than corrupt dates by ~50,000 years. No effect on time-typed columns receiving time.Time/time.Duration Go values, which carry their own unit unambiguously, and no effect on non-time columns. Requires schema_metadata to be set.

Type: bool

Default: false

schema_evolution.schema_metadata

The name of a message metadata field containing a schema definition. When set, the schema is used to determine column types during schema evolution and table creation instead of inferring types from values. The schema must be in the standard common schema format (the same format used by the parquet_encode processor’s schema_metadata field). For batches of messages, the first message’s schema is used. Record presence drives schema shape: fields declared in the schema metadata that are absent from the record are not added to the table, while the metadata controls column ordering, naming, and types for fields that are present. In case-insensitive mode, top-level column names use the metadata’s casing; record keys are matched by case-folding and the metadata’s name is what lands in the table.

Type: string

Default: ""

schema_evolution.table_location

A prefix used as the location for new tables when the catalog does not automatically assign one. For example, AWS Glue requires explicit table locations. When set, table locations are derived by appending the namespace and table name to the prefix, in the form <prefix><namespace>/<table>.

Type: string

# Examples:
table_location: s3://my-iceberg-bucket/

storage

Storage backend configuration for data files. Exactly one of aws_s3, gcp_cloud_storage, or azure_blob_storage must be specified.

Type: object

storage.aws_s3

S3 storage configuration.

Type: object

storage.aws_s3.bucket

The S3 bucket name.

Type: string

# Examples:
bucket: my-iceberg-data

storage.aws_s3.credentials

Static AWS credentials for S3 access. When not specified, credentials are loaded from the default AWS credential chain.

Type: object

storage.aws_s3.credentials.id

The AWS access key ID.

Type: string

storage.aws_s3.credentials.secret

The AWS secret access key.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

storage.aws_s3.credentials.token

The AWS session token, required when using short term credentials.

Type: string

storage.aws_s3.endpoint

Custom endpoint for S3-compatible storage (for example, MinIO).

Type: string

# Examples:
endpoint: http://localhost:9000

storage.aws_s3.force_path_style_urls

Forces the client API to use path style URLs, which is often required when connecting to custom endpoints.

Type: bool

Default: false

storage.aws_s3.region

The AWS region.

Type: string

# Examples:
region: us-west-2

storage.azure_blob_storage

Azure Blob Storage (ADLS Gen2) configuration.

Type: object

storage.azure_blob_storage.container

The Azure blob container name.

Type: string

# Examples:
container: iceberg-data

storage.azure_blob_storage.endpoint

Custom endpoint for Azure-compatible storage.

Type: string

storage.azure_blob_storage.storage_access_key

Azure storage access key for shared key authentication.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

storage.azure_blob_storage.storage_account

The Azure storage account name.

Type: string

# Examples:
storage_account: mystorageaccount

storage.azure_blob_storage.storage_connection_string

Azure storage connection string. Use this or other auth methods, not both.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

storage.azure_blob_storage.storage_sas_token

SAS token for authentication. Prefix with the container name followed by a dot if container-specific.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

storage.gcp_cloud_storage

Google Cloud Storage configuration.

Type: object

storage.gcp_cloud_storage.bucket

The GCS bucket name.

Type: string

# Examples:
bucket: my-iceberg-data

storage.gcp_cloud_storage.credentials_file

Path to a GCP credentials JSON file.

Type: string

storage.gcp_cloud_storage.credentials_json

GCP credentials JSON content. Use this or credentials_file, not both.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

storage.gcp_cloud_storage.credentials_type

The type of credentials to use. Valid values: service_account, authorized_user, impersonated_service_account, external_account.

Type: string

# Examples:
credentials_type: service_account

storage.gcp_cloud_storage.endpoint

Custom endpoint for GCS-compatible storage.

Type: string

table

The destination Iceberg table name.

This field supports interpolation functions.

Type: string

# Examples:
table: user_events

# ---

table: events_${!meta("topic")}

Examples

Change-data-capture upsert/delete

Materialize a change-data-capture stream into an Iceberg table. A mapping derives the row operation from the source’s operation field (here Debezium’s op: c`reate / `r`ead / `u`pdate map to `upsert, d`elete maps to `delete) and selects the row image, while identifier_fields is the primary key. Note that creates map to upsert, never insert, so re-delivered or snapshot rows do not duplicate.

input:
  redpanda:
    seed_brokers: [ localhost:9092 ]
    topics: [ dbserver.inventory.customers ]
    consumer_group: iceberg_sink

pipeline:
  processors:
    - mapping: |
        meta op = match this.op {
          "d" => "delete",
          _   => "upsert",
        }
        # Debezium puts the row image in 'after' (or 'before' for deletes).
        root = this.after | this.before

output:
  iceberg:
    catalog:
      url: http://localhost:8181/api/catalog
    namespace: inventory
    table: customers
    row_operation: ${! metadata("op") }
    identifier_fields: [ id ]
    # Keyed writes must stay ordered: a single batch in flight prevents
    # concurrent batches from committing a stale update over a newer one.
    max_in_flight: 1
    storage:
      aws_s3:
        bucket: my-iceberg-data
        region: us-east-1

CDC into a Snowflake- or Databricks-readable table (copy-on-write)

Materialize a change-data-capture stream into an Iceberg table that must be read by an engine-backed catalog such as Snowflake or the Databricks Unity Catalog, which cannot read merge-on-read equality deletes. merge_strategy: copy-on-write rewrites whole data files so the table only ever holds plain data files that every engine can read. As with any keyed workload it requires max_in_flight: 1; because copy-on-write rewrites every file containing a touched key, prefer large batches and a table sorted by the identifier key, and run regular compaction and snapshot expiry.

input:
  redpanda:
    seed_brokers: [ localhost:9092 ]
    topics: [ dbserver.inventory.customers ]
    consumer_group: iceberg_sink_cow

pipeline:
  processors:
    - mapping: |
        meta op = match this.op {
          "d" => "delete",
          _   => "upsert",
        }
        root = this.after | this.before

output:
  iceberg:
    catalog:
      url: http://localhost:8181/api/catalog
    namespace: inventory
    table: customers
    row_operation: ${! metadata("op") }
    identifier_fields: [ id ]
    # copy-on-write produces only plain data files, so Snowflake and the
    # Databricks Unity Catalog can read the result (unlike equality deletes).
    merge_strategy: copy-on-write
    # Keyed writes must stay ordered; a single batch in flight prevents a stale
    # update from overwriting a newer one for the same key.
    max_in_flight: 1
    # Amortise the file rewrites over larger commits.
    batching:
      count: 5000
      period: 30s
    storage:
      aws_s3:
        bucket: my-iceberg-data
        region: us-east-1