dlt-hub / dlt-hub/dlt

feat: Stream Iceberg writes with constant memory and atomic commits

Open
#3,752 0 comments 0 reactions 1 assignee Claimed by @rudolfix View on GitHub
destination enhancement needs design
Dominant language
Python
Stars
5.9k
Forks
600
Avg merge
1d 14h
Merged PRs (30d)
38

Description

## Feature description

Stream all Iceberg write paths (append, replace, upsert, insert-only) batch-by-batch with constant memory and minimal Iceberg snapshots.

Currently, when loading data into Iceberg tables via the `filesystem` destination, the entire Arrow dataset is materialised in memory before writing. Specifically, `IcebergLoadFilesystemJob.run()` calls `self.arrow_dataset.to_table()`, which reads all parquet files into a single in-memory Arrow table. For a dataset with 1M+ rows and hundreds of columns, this alone can consume several gigabytes of RAM and cause OOM in constrained environments.

Additionally, during upserts, `merge_iceberg_table` calls `table.upsert()` once per batch. Each call produces a separate Iceberg snapshot. For large datasets split into many batches, this leads to hundreds of snapshots, which degrades table scan performance and increases catalog metadata overhead.

The Delta Lake code path in dlt already streams via `RecordBatchReader` (see `DeltaLoadFilesystemJob`), so this would bring Iceberg to parity with Delta.

## Are you a dlt user?

Yes.

## Use case

Loading large datasets into Iceberg tables via the filesystem destination without running out of memory, and producing minimal Iceberg snapshots for all write dispositions (append, replace, upsert, insert-only).

The current implementation requires the entire dataset to fit in RAM, which makes it unusable for large-scale production workloads in memory-constrained environments. There is no way to load a dataset larger than available RAM into an Iceberg table through dlt today.

## Proposed solution

Accept `pa.RecordBatchReader` (in addition to `pa.Table`) in `write_iceberg_table` and `merge_iceberg_table` to enable batch-by-batch streaming with constant memory:

- **Append / Replace**: Stream batches as individual parquet files uploaded via Iceberg's IO, then register all files atomically with `table.add_files()` (append) or `delete` + `add_files` in a single transaction (replace). One atomic commit for the entire load.

- **Upsert / Insert-only**: Use a single `table.transaction()` — updates go via `txn.overwrite()` only when rows actually changed, inserts are collected as remote parquet files and registered via `txn.add_files()` at the end of the transaction. This produces minimal snapshots (typically 1 OVERWRITE + 1 APPEND) instead of one snapshot per batch. For partitioned tables, inserts are handled in a separate transaction using `txn.append()` (since pyiceberg 0.9.x crashes when mixing `overwrite` + `append` in one transaction), resulting in 2 snapshots instead of 1. This can be collapsed into a single transaction once pyiceberg fixes this upstream.

- **IcebergLoadFilesystemJob**: Read schema from parquet file header (`pq.read_schema`) instead of materialising the dataset. Stream via `RecordBatchReader` for all code paths. Explicit `gc.collect()` after loads to release memory promptly.

- **Filesystem destination**: Set `recommended_file_size = 128MB` so upstream produces reasonably-sized parquet files for streaming.

This is the same approach recommended by a pyiceberg maintainer (@kevinjqliu) in [apache/iceberg-python#2152](https://github.com/apache/iceberg-python/issues/2152), referencing [iceberg-go's implementation](https://github.com/apache/iceberg-go/pull/369): materialize arrow streams as individual parquet files, then register them via `add_files`.

Since pyiceberg doesn't natively support streaming writes yet, this handles it at the dlt layer — streaming `RecordBatch`es one at a time, writing each to a temp parquet file, uploading via Iceberg's IO, and registering all files atomically. Memory stays constant regardless of dataset size while maintaining Iceberg transactional guarantees (single atomic commit for all paths except partitioned upserts, which require two transactions due to a pyiceberg 0.9.x limitation).

I have a working implementation and will open a PR.

## Related issues

Upstream pyiceberg issues describing the same limitation:

- [apache/iceberg-python#1004](https://github.com/apache/iceberg-python/issues/1004) — OOM during `table.append()` due to full materialization. The issue author noted: *"it is not a bug, when you do arrow(), all the data will be loaded into memory… either you add more ram, or make catalog.load_table accept recordbatch as a valid input (like delta-rs)"*

- [apache/iceberg-python#2152](https://github.com/apache/iceberg-python/issues/2152) — Open feature request for `RecordBatchReader` / `Scanner` write support. A pyiceberg maintainer acknowledged it's hard to do without full materialization and suggested using Spark for large datasets. However, another maintainer (@kevinjqliu) pointed to [iceberg-go's approach](https://github.com/apache/iceberg-go/pull/369) — which is exactly what this implementation does.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.