Skip to main content

iceberg_parquet processor

Encodes a batch of JSON messages into a single Apache Iceberg-ready Parquet file.

# Config fields, showing default values
pipeline:
  processors:
    - label: ""
      iceberg_parquet:
        schema: [] # No default (required)
        compression: "zstd"
        schema_metadata_key: "@iceberg_schema"

Converts the typical JSON a pipeline produces into the columnar Parquet that Apache Iceberg tables are built from, so the data lands query-ready for Spark, Trino, Snowflake, Athena and DuckDB instead of as raw JSON.

Designed for the batching.processors slot of an object-storage output (for example aws_s3 or gcp_cloud_storage): the output collects a batch, this processor turns the whole batch into one Parquet file, and the output writes that file. The batch size and period configured on the output therefore set the file grain.

The file is written to be registered into an Iceberg table by a central process running add_files (name-mapping), so the edge node never needs to reach a catalog. To make that registration deterministic, the derived Iceberg table schema is attached to the output message as metadata (see schema_metadata_key) — this resolves the Parquet-to-Iceberg type ambiguity a committer would otherwise have to guess at.

Timestamps are always encoded at microsecond precision; nanosecond Parquet timestamps are not readable by Spark, Athena or DuckDB, so that footgun is removed by construction. Columns absent from a message are written as null (mark them optional); fields present in the message but absent from the schema are dropped.

Examples​

Land JSON as Iceberg-ready Parquet in object storage​

An aws_s3 output batches messages; this processor encodes each batch as one Iceberg-ready Parquet file under a partitioned prefix, and a central process registers the files into the table with add_files.

output:
  aws_s3:
    bucket: my-lakehouse-bucket
    path: warehouse/analytics/events/dt=${! now().ts_strftime("%Y-%m-%d") }/data-${! uuid_v4() }.parquet
    batching:
      count: 5000
      period: 60s
      processors:
        - iceberg_parquet:
            compression: zstd
            schema:
              - { name: id, type: string }
              - { name: event_time, type: timestamptz }
              - { name: reading, type: long }
              - { name: status, type: string, optional: true }

Fields​

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 (non-optional) column whose value is missing in a message fails the batch.

Type: bool
Default: false

compression​

The Parquet compression codec applied to every column.

Type: string
Default: "zstd"

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

schema_metadata_key​

Metadata key on the emitted file message carrying the derived Iceberg table schema as JSON, for a central add_files committer to create or evolve the table. Set to an empty string to skip attaching it.

Type: string
Default: "@iceberg_schema"