Skip to main content

iceberg_api output

Writes a batch of JSON messages to an Apache Iceberg table and commits the snapshot to a REST catalog.

# Common config fields, showing default values
output:
  label: ""
  iceberg_api:
    catalog:
      url: "" # No default (required)
      warehouse: "" # No default (optional)
      auth:
        oauth2:
          server_uri: "" # No default (optional)
          client_id: "" # No default (optional)
          client_secret: "" # No default (optional)
          scope: "" # No default (optional)
        bearer: "" # No default (optional)
        aws_sigv4:
          region: "" # No default (optional)
          service: "execute-api"
      tls_skip_verify: false
      headers: {} # No default (optional)
    namespace: "" # No default (required)
    table: "" # No default (required)
    schema: [] # No default (required)
    compression: "zstd"
    create_table_if_missing: false
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""

Converts each batch of JSON into columnar data, writes it to the table's object storage, and commits a new snapshot to the Iceberg REST catalog in one transaction. The table is query-ready immediately for Spark, Trino, Snowflake, Athena and DuckDB, with no separate file-registration step.

The node must be able to reach the REST catalog. For air-gapped or very high-fan-in fleets, prefer the iceberg_parquet processor (which stages Parquet to object storage for a central committer) instead.

The output's batching count/period is the commit grain: one snapshot per batch. Commits are serialized (one in flight) because concurrent commits to one table are optimistic-concurrency conflicts; at fleet scale, route different nodes or networks to different tables or namespaces (for example via an environment variable in table).

Timestamps are always encoded at microsecond precision (nanosecond Parquet is not readable by Spark, Athena or DuckDB). Columns absent from a message are written as null (mark them optional); fields present in a message but absent from the schema are dropped. This release supports inserts only.

Fields​

catalog​

Iceberg REST catalog connection.

Type: object

catalog.url​

The Iceberg REST catalog base URL, e.g. https://polaris.example.com/api/catalog.

Type: string

catalog.warehouse​

The catalog warehouse / prefix, when the catalog requires one.

Type: string

catalog.auth​

Catalog authentication. Set at most one of oauth2, bearer, or aws_sigv4; omit for an unauthenticated catalog.

Type: object

catalog.auth.oauth2​

OAuth2 client-credentials auth.

Type: object

catalog.auth.oauth2.server_uri​

The OAuth2 token endpoint, absolute or relative to the catalog URL. When unset, the catalog client's default token endpoint is used.

Type: string

catalog.auth.oauth2.client_id​

OAuth2 client id.

Type: string

catalog.auth.oauth2.client_secret​

OAuth2 client secret.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

catalog.auth.oauth2.scope​

OAuth2 scope.

Type: string

catalog.auth.bearer​

A static bearer token, as an alternative to oauth2.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

catalog.auth.aws_sigv4​

AWS SigV4 request signing (e.g. for an API Gateway-fronted catalog).

Type: object

catalog.auth.aws_sigv4.region​

AWS region for SigV4 signing.

Type: string

catalog.auth.aws_sigv4.service​

AWS service name for SigV4 signing.

Type: string
Default: "execute-api"

catalog.tls_skip_verify​

Skip TLS certificate verification for the catalog connection. Do not enable in production.

Type: bool
Default: false

catalog.headers​

Extra HTTP headers to send on every catalog request.

Type: map of string

namespace​

The table namespace (database). Supports environment-variable interpolation at startup, e.g. analytics or db_${NETWORK_ID}, so different networks can target different namespaces.

Type: string

table​

The table name. Supports environment-variable interpolation at startup, e.g. events or events_${NODE_ID}, so different nodes can target different tables to avoid commit contention.

Type: string

schema​

The Iceberg table columns, in order. Only top-level scalar columns are supported in this release.

Type: array of object

schema[].name​

The column name. Matched against the top-level JSON field of the same name.

Type: string

schema[].type​

The Iceberg column type. timestamptz is adjusted to UTC and encoded at microsecond precision, from an RFC3339 string or a time value.

Type: string

Options: boolean, int, long, float, double, string, timestamptz

schema[].optional​

Whether the column is nullable. A required column whose value is missing in a message fails the batch.

Type: bool
Default: false

compression​

The Parquet compression codec, set as a write property on tables this output creates. An existing table keeps the codec recorded in its own metadata.

Type: string
Default: "zstd"

Options: uncompressed, snappy, gzip, brotli, zstd, lz4raw

create_table_if_missing​

Create the namespace and table if they do not exist, using the declared schema. When false, the table must already exist in the catalog.

Type: bool
Default: false

batching​

Allows you to configure a batching policy.

Type: object

batching.count​

A number of messages at which the batch should be flushed. If 0 disables count based batching.

Type: int
Default: 0

batching.byte_size​

An amount of bytes at which the batch should be flushed. If 0 disables size based batching.

Type: int
Default: 0

batching.period​

A period in which an incomplete batch should be flushed regardless of its size.

Type: string
Default: ""

batching.check​

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

Type: string
Default: ""

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. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.

Type: array of processor