iceberg-loader

📚 Documentation

iceberg-loader wraps PyIceberg to load data into Apache Iceberg tables. It uses PyArrow and handles mixed JSON fields, schema evolution, idempotent replacement, upserts, batching, and streaming.

Status: Actively developed and under testing. Tested against Hive Metastore and REST Catalog (Tabular, Polaris, self-hosted).

Why iceberg-loader?

• Mixed JSON fields: converts dict and list values to strings before writing, including fields with mixed types.
• Schema evolution: adds columns when enabled and preserves field IDs.
• Write modes: append, overwrite, idempotent replacement via replace_filter, and upsert.
• Streaming: commit intervals, batches, and IPC streams.
• Configuration: LoaderConfig sets defaults that can be overridden per call.

Install

pip install "iceberg-loader[all]"

Or with uv:

uv add "iceberg-loader[all]"

Quickstart

Hive Metastore

from pyiceberg.catalog import load_catalog
from iceberg_loader import LoaderConfig, load_data_to_iceberg
from iceberg_loader.utils.arrow import create_arrow_table_from_data

catalog = load_catalog("default")
table_id = ("default", "comparison_complex_json")

data = [
    {"id": 1, "complex_field": {"a": 1, "b": "nested"}, "signup_date": "2023-01-01"},
    {"id": 2, "complex_field": {"a": 2, "b": "another", "c": [1, 2]}, "signup_date": "2023-01-02"},
    {"id": 3, "complex_field": [1, 2, 3], "signup_date": "2023-01-02"},
]

arrow_table = create_arrow_table_from_data(data)

config = LoaderConfig(write_mode="append", partition_col="day(signup_date)", schema_evolution=True)
load_data_to_iceberg(arrow_table, table_id, catalog, config=config)

REST Catalog

from iceberg_loader import LoaderConfig, get_rest_catalog, load_data_to_iceberg
from iceberg_loader.utils.arrow import create_arrow_table_from_data

catalog = get_rest_catalog(
    uri="https://api.tabular.io/ws",
    warehouse="s3://my-bucket/warehouse/",
    credential="your-oauth2-credential",
)
table_id = ("default", "my_table")

data = [
    {"id": 1, "name": "Alice", "signup_date": "2023-01-01"},
    {"id": 2, "name": "Bob", "signup_date": "2023-01-02"},
]

arrow_table = create_arrow_table_from_data(data)

config = LoaderConfig(write_mode="append", partition_col="day(signup_date)", schema_evolution=True)
load_data_to_iceberg(arrow_table, table_id, catalog, config=config)

Or use pyiceberg.yaml to configure your REST catalog and load it by name:

# ~/.pyiceberg.yaml (see examples/pyiceberg.yaml.sample for a full annotated example)
catalog:
  my-rest-catalog:
    type: rest
    uri: https://your-catalog-server.com/api
    warehouse: s3://my-bucket/warehouse/
    credential: your-oauth2-credential
    s3.endpoint: https://s3.amazonaws.com
    s3.region: us-east-1

from pyiceberg.catalog import load_catalog

catalog = load_catalog("my-rest-catalog")
# ... rest is identical to above

Which function to use?

Preparing Data

These helpers convert Python dictionaries to Arrow data, including mixed fields:

from iceberg_loader.utils.arrow import create_arrow_table_from_data, create_record_batches_from_dicts

# 1. Convert list of dicts -> pa.Table
arrow_table = create_arrow_table_from_data(data_list)

# 2. Convert iterator of dicts -> Iterator[pa.RecordBatch]
batches = create_record_batches_from_dicts(data_generator(), batch_size=10000)

Alternatively, use standard PyArrow conversion: pa.Table.from_pylist(data).

The helpers serialize dicts and lists as JSON and convert other non-null values to strings too. For example, an integer id becomes an Arrow string. If you need to preserve scalar types, prepare a typed Arrow table with PyArrow instead.

For timestamp columns, prefer partition transforms such as day(ts) or hour(ts), especially when using load_timestamp.

Public API & Stability

• Top-level exports include LoaderConfig, IcebergLoader, load_data_to_iceberg, load_batches_to_iceberg, load_ipc_stream_to_iceberg, get_rest_catalog, and expire_snapshots.
• Data conversion helpers are available from iceberg_loader.utils.arrow. Pass loading options via LoaderConfig.
• Avoid legacy positional arguments—use the config parameter only.
• LoaderConfig validates partition expressions and rejects unsafe combos (e.g., replace_filter with upsert, identity partition on _load_dttm).

How we version

• Semantic Versioning starting at 0.1.x: MINOR for compatible features, PATCH for fixes, MAJOR for breaking API changes.
• Breaking changes to documented APIs are noted in RELEASE.md.

Release checklist

• Bump version in pyproject.toml and src/iceberg_loader/__about__.py (they must match).
• Update RELEASE.md with highlights and breaking notes.
• Run uv lock, commit uv.lock if it changes, then verify it with uv lock --locked.
• Run uv run ruff check ., uv run ty check, and uv run python -m pytest.
• Tag and push (git tag -a vX.Y.Z ...), then let CI publish.

Contributing

See CONTRIBUTING.md for setup, coding style, and PR guidelines.

uv run ruff check . && uv run ruff format --check . && uv run ty check
uv run pytest

Contributors

Made with contrib.rocks.

License

iceberg-loader is distributed under the terms of the MIT license.