dlt-iceberg

A dlt destination for Apache Iceberg tables using REST catalogs.

Features

• Atomic Multi-File Commits: Multiple parquet files committed as single Iceberg snapshot per table
• REST Catalog Support: Works with Nessie, Polaris, AWS Glue, Unity Catalog
• Credential Vending: Most REST catalogs vend storage credentials automatically
• Partitioning: Full support for Iceberg partition transforms via iceberg_adapter()
• Merge Strategies: Delete-insert and upsert with hard delete support
• DuckDB Integration: Query loaded data via pipeline.dataset()
• Schema Evolution: Automatic schema updates when adding columns

Installation

pip install dlt-iceberg

Or with uv:

uv add dlt-iceberg

Quick Start

import dlt
from dlticeberg import icebergrest
@dlt.resource(name="events", write_disposition="append") def generate_events():     yield {"event_id": 1, "value": 100}
pipeline = dlt.pipeline(     pipelinename="mypipeline",     destination=iceberg_rest(         catalog_uri="https://my-catalog.example.com/api/catalog",         namespace="analytics",         warehouse="my_warehouse",         credential="client-id:client-secret",         oauth2serveruri="https://my-catalog.example.com/oauth/tokens",     ), )
pipeline.run(generate_events())

Query Loaded Data

# Query data via DuckDB
dataset = pipeline.dataset()
Access as dataframe
df = dataset["events"].df()
Run SQL queries
result = dataset.query("SELECT * FROM events WHERE value > 50").fetchall()
Get Arrow table
arrow_table = dataset["events"].arrow()

Merge/Upsert

@dlt.resource(
    name="users",
    write_disposition="merge",
    primarykey="userid"
)
def generate_users():
    yield {"user_id": 1, "name": "Alice", "status": "active"}
pipeline.run(generate_users())

Configuration

Required Options

iceberg_rest(
    catalog_uri="...",    # REST catalog endpoint (or sqlite:// for local)
    namespace="...",      # Iceberg namespace (database)
)

Authentication

Choose based on your catalog:

| Catalog | Auth Method | |---------|-------------| | Polaris, Lakekeeper | credential + oauth2serveruri | | Unity Catalog | token | | AWS Glue | sigv4enabled + signingregion | | Local SQLite | None needed |

Most REST catalogs (Polaris, Lakekeeper, etc.) vend storage credentials automatically via the catalog API. You typically don't need to configure S3/GCS/Azure credentials manually.

iceberg_rest(
    # ... required options ...
# Manual storage credentials (usually not needed with credential vending)     s3_endpoint="...",     s3accesskey_id="...",     s3secretaccess_key="...",     s3_region="...",
# Performance tuning     max_retries=5,               # Retry attempts for transient failures     retrybackoffbase=2.0,      # Exponential backoff multiplier     mergebatchsize=500000,     # Rows per batch for merge operations     strict_casting=False,        # Fail on potential data loss
# Table management     tablelocationlayout=None,  # Custom table location pattern     registernewtables=False,   # Register tables found in storage     harddeletecolumn="dltdeleted_at",  # Column for hard deletes )

Catalog Examples

iceberg_rest(
    catalog_uri="http://localhost:8282/catalog/",
    warehouse="test-warehouse",
    namespace="my_namespace",
    s3_endpoint="http://localhost:9000",
    s3accesskey_id="minioadmin",
    s3secretaccess_key="minioadmin",
    s3_region="us-east-1",
)

iceberg_rest(
    catalog_uri="https://polaris.example.com/api/catalog",
    warehouse="my_warehouse",
    namespace="production",
    credential="client-id:client-secret",
    oauth2serveruri="https://polaris.example.com/api/catalog/v1/oauth/tokens",
)

iceberg_rest(
    catalog_uri="https://<workspace>.cloud.databricks.com/api/2.1/unity-catalog/iceberg-rest",
    warehouse="<catalog-name>",
    namespace="<schema-name>",
    token="<databricks-token>",
)

iceberg_rest(
    catalog_uri="https://glue.us-east-1.amazonaws.com/iceberg",
    warehouse="<account-id>:s3tablescatalog/<bucket>",
    namespace="my_database",
    sigv4_enabled=True,
    signing_region="us-east-1",
)

iceberg_rest(
    catalog_uri="sqlite:///catalog.db",
    warehouse="file:///path/to/warehouse",
    namespace="my_namespace",
)

iceberg_rest(
    catalog_uri="http://localhost:19120/iceberg/main",
    namespace="my_namespace",
    s3_endpoint="http://localhost:9000",
    s3accesskey_id="minioadmin",
    s3secretaccess_key="minioadmin",
    s3_region="us-east-1",
)

Partitioning

Using iceberg_adapter (Recommended)

The iceberg_adapter function provides a clean API for configuring Iceberg partitioning:

from dlticeberg import icebergadapter, iceberg_partition
@dlt.resource(name="events") def events():     yield {"eventdate": "2024-01-01", "userid": 123, "region": "US"}
Single partition
adapted = iceberg_adapter(events, partition="region")
Multiple partitions with transforms
adapted = iceberg_adapter(
    events,
    partition=[
        icebergpartition.month("eventdate"),
        icebergpartition.bucket(10, "userid"),
        "region",  # identity partition
    ]
)
pipeline.run(adapted)

Partition Transforms

# Temporal transforms (for timestamp/date columns)
icebergpartition.year("createdat")
icebergpartition.month("createdat")
icebergpartition.day("createdat")
icebergpartition.hour("createdat")
Identity (no transformation)
iceberg_partition.identity("region")
Bucket (hash into N buckets)
icebergpartition.bucket(10, "userid")
Truncate (truncate to width)
iceberg_partition.truncate(4, "email")
Custom partition field names
icebergpartition.month("createdat", "event_month")
icebergpartition.bucket(8, "userid", "user_bucket")

Using Column Hints

Prefer iceberg_adapter for partitioning. If you need to set column hints directly, use the x-partition custom hints; raw partition_transform fields are rejected by dlt schema validation.

@dlt.resource(
    name="events",
    columns={
        "event_date": {
            "data_type": "date",
            "x-partition": True,
            "x-partition-transform": "day",
        },
        "user_id": {
            "data_type": "bigint",
            "x-partition": True,
            "x-partition-transform": "bucket[10]",
        }
    }
)
def events():
    ...

Write Dispositions

Append

write_disposition="append"

Replace

write_disposition="replace"

Merge

Delete-Insert Strategy (Default)

@dlt.resource(
    write_disposition={"disposition": "merge", "strategy": "delete-insert"},
    primarykey="userid"
)

Upsert Strategy

@dlt.resource(
    write_disposition={"disposition": "merge", "strategy": "upsert"},
    primarykey="userid"
)

Hard Deletes

Mark rows for deletion by setting the dltdeleted_at column:

@dlt.resource(
    write_disposition={"disposition": "merge", "strategy": "delete-insert"},
    primarykey="userid"
)
def userswithdeletes():
    from datetime import datetime
    yield {"userid": 1, "name": "alice", "dltdeletedat": None}  # Keep
    yield {"userid": 2, "name": "bob", "dltdeletedat": datetime.now()}  # Delete

Development

Run Tests

# Start Docker services (for Nessie tests)
docker compose up -d
Run all tests
uv run pytest tests/ -v
Run only unit tests (no Docker required)
uv run pytest tests/ --ignore=tests/nessie -v
Run Nessie integration tests
uv run pytest tests/nessie/ -v

Project Structure

dlt-iceberg/
├── src/dlt_iceberg/
│   ├── init.py           # Public API
│   ├── destination_client.py # Class-based destination (atomic commits)
│   ├── destination.py        # Function-based destination (legacy)
│   ├── adapter.py            # iceberg_adapter() for partitioning
│   ├── sql_client.py         # DuckDB integration for dataset()
│   ├── schema_converter.py   # dlt → Iceberg schema conversion
│   ├── schema_casting.py     # Arrow table casting
│   ├── schema_evolution.py   # Schema updates
│   ├── partition_builder.py  # Partition specs
│   └── error_handling.py     # Retry logic
├── tests/
│   ├── testadapter.py       # icebergadapter tests
│   ├── test_capabilities.py  # Hard delete, partition names tests
│   ├── test_dataset.py       # DuckDB integration tests
│   ├── testmergedisposition.py
│   ├── testschemaevolution.py
│   └── ...
├── examples/
│   ├── incremental_load.py   # CSV incremental loading
│   ├── merge_load.py         # CSV merge/upsert
│   └── data/                 # Sample CSV files
└── docker-compose.yml        # Nessie + MinIO for testing

How It Works

The class-based destination uses dlt's JobClientBase interface to accumulate parquet files during a load and commit them atomically in complete_load():

• dlt extracts data and writes parquet files
• Each file is registered in module-level global state
• After all files complete, complete_load() is called
• All files for a table are combined and committed as single Iceberg snapshot
• Each table gets one snapshot per load

License

MIT License - see LICENSE file

Resources

• dlt Documentation
• Apache Iceberg
• PyIceberg
• Iceberg REST Spec