Apache Iceberg

Writes records as Parquet files to an Iceberg table in S3, or a store that speaks its API, and commits them to the table through its REST catalog. Each commit is a snapshot of the table.

Example

Signs in to a Polaris catalog with OAuth client credentials, the client's secret in the worker-host variable CATALOG_CLIENT_SECRET, and writes the table analytics.web.access_log, partitioned by the day of ts and 16 buckets of user.id when it creates it. Storage is signed with what the catalog vends.

{
  "id": "access-log",
  "type": "iceberg",
  "catalog": "https://polaris.example.com/api/catalog",
  "warehouse": "analytics",
  "namespace": "analytics.web",
  "table": "access_log",
  "partitionBy": [
    "day(ts)",
    "bucket(16, user.id)"
  ],
  "clientId": "smeltr",
  "clientSecret": "CATALOG_CLIENT_SECRET",
  "scope": "PRINCIPAL_ROLE:ALL"
}

Open complete screenshot ↗ Open complete screenshot ↗

Signs in to AWS Glue's Iceberg endpoint with AWS signatures, as the worker host's AWS credentials, in the region the address names. Set the warehouse to the account's id, and the namespace to a Glue database.

{
  "id": "access-log",
  "type": "iceberg",
  "catalog": "https://glue.us-east-2.amazonaws.com/iceberg",
  "warehouse": "111122223333",
  "namespace": "web",
  "table": "access_log",
  "signingName": "glue"
}

Before you start

• Have an Iceberg REST catalog the worker hosts can reach, such as Apache Polaris, AWS Glue's Iceberg endpoint, or S3 Tables. The table's files are in S3 or a store that speaks its API.
• Create the namespace in the catalog. The destination creates a missing table in it, but not the namespace.
• A table that exists is of format version 2 and not encrypted. A table the destination creates is of version 2.
• Give the principal the destination signs in to the catalog as permission to load and update the table, and to create it when it is missing. Signing in lists the ways.
• Give whatever signs for storage permission to write and delete objects under the table's data location, and to read and write objects under its metadata location: the credentials the catalog vends, the access key or role the destination names, or the worker host's own. The destination deletes no metadata.
• Keep the table's location, its write.data.path and its write.metadata.path in one bucket. The destination refuses to start on a table whose paths are in another bucket than its location.
• Send this destination records of one type: the routes writing to it write the same record, which is the table's schema. See Columns.

Settings

It also takes the settings every destination takes.

Advanced: Catalog and storage

Advanced: Delivery

Signing in

The destination signs in to the catalog one way, and to the table's storage another. The GUI form's catalog sign-in chooses the first.

Storage is signed with the first of these the destination has:

1. The access key in accessKeyId and secretAccessKey, or the role in roleArn.
2. The credentials the catalog vends for the table. The destination asks for them when it loads the table, and asks again five minutes before they expire.
3. The worker host's own AWS credentials, such as an instance role.

The table's bucket is in region, else the region the catalog gives the table, else the one the catalog's AWS address names, else the worker host's, else us-east-1. It is reached at endpoint, else the endpoint the catalog gives the table, else, with no access key named, the one the worker host's AWS settings name, else AWS's own address for the region.

Columns

The table's schema is the record the routes write here, its fields the table's columns, nested fields as structs and arrays as lists. A missing table is created with that schema and with partitionBy as its partition spec. A table that exists keeps its columns and its spec: a field it lacks, at any depth, is added as an optional column before any file is written, and a column the record lacks reads as null in the rows written. A field that can hold no value, null or a list or closed struct of nothing else, has no column. Each field is written as the Iceberg type of its record type:

Every column the destination writes is optional. It refuses to start when the table holds a field as another type, or has a required column the record may leave null or does not write.

How it works

When it starts, the destination loads the table from the catalog, or creates it, and adds the record's new columns. It then writes each batch's records into Parquet files under the table's data location, write.data.path when the table sets it, partitioned by the table's own spec: identity, year, month, day, hour, bucket and truncate. Each file carries the table's column ids. A file stays open across batches, one per partition, at most 64 at once; a 65th closes the one written least recently. A file also closes at maxObjectBytes.

The destination commits when fileIntervalSecs passes, when the records it holds near bufferBytes, and whenever the worker waits for its destinations to acknowledge everything they hold. A commit closes every open file, writes a manifest of the new files and a manifest list under the table's metadata location, merging the parent's small manifests as the table's commit.manifest.* properties say, and asks the catalog to append them as the table's next snapshot on top of the one it read. A batch is acknowledged once the commit holding its records lands.

A commit another writer beat to the table is made again at once on top of the new snapshot, up to 8 times. A failed call is tried again with backoff until stallAfterSecs. When the catalog's answer to a commit is lost, the destination looks for that commit in the table before it tries again, so it does not make it twice. Past stallAfterSecs the destination aborts its open uploads and writes the batches again. It deletes the files it finished unless a lost answer may have committed them; then they stay, and their rows may be in the table twice. A file storage keeps after a failure is an orphan for the table's own cleanup.

Each commit is a snapshot the table keeps until it is expired. Expire snapshots and compact small files with the table's own maintenance.

A file larger than 8 MiB goes up as a multipart upload, which the destination completes when the file closes and aborts when it fails. Two things leave one unfinished: the worker stops without shutting down (it is killed, it crashes, or its host is lost) while the file is open, or storage does not answer the abort within stallAfterSecs. A file stays open at most fileIntervalSecs, so give the bucket a lifecycle rule that aborts incomplete multipart uploads after a day: longer than any file stays open.

Verify

Send a record with a unique marker through the source. After fileIntervalSecs, look for a new snapshot in the table's history, and query the table for the marker with any engine that reads Iceberg.

Troubleshooting

the catalog refused …

The catalog answered with the status the message names. On 401 or 403, check the sign-in: the token, the client's id, secret and scope, or the AWS credentials and the signing name. On 404, check catalog, warehouse and that the namespace exists.

the routes writing here write different records, and a table has one schema

Two routes feeding the destination write records of different types. Make them write the same record, or give each table a destination of its own.

… is … in the table, and the record writes … there as …

The table holds a column as another type than the record's field. Rename the field in the pipeline or the column in the table.

the table's column … is required …

Every column the destination writes is optional. Make the column optional in the table, or send the records to a table of their own.

it is of format version …, and only version 2 is written

Write to a table of format version 2.

it is encrypted, and an encrypted table is not written

Write to a table that is not encrypted.

the catalog vended no single credential for both the table's data location …

The catalog's credentials cover one of the table's locations and not the other. Name an access key or a role for storage, or have the catalog vend one credential for both.

the catalog places the table's storage at …, plain HTTP off this host

The catalog gives the table an http:// endpoint on another host. Name an https:// endpoint.

Access denied writing to storage

Check that whatever signs for storage may write and delete objects under the table's data location, and read and write objects under its metadata location.

See also

• Destinations
• Amazon S3 destination
• Delivery
• Secrets & credentials