pyiceberg
PyIceberg is the Apache implementation of the Iceberg table format in Python, with no JVM and no Spark anywhere in the picture. Everything goes through a catalog: you point load_catalog at a REST endpoint, AWS Glue, Hive, a SQL database or DynamoDB, load a table by name, and from there you can read the metadata, plan a scan and read data. A scan is where the format earns its keep, because the manifest files carry per-file column statistics, so a row filter can eliminate whole Parquet files before any of them are opened. What comes back is an Arrow table, a pandas DataFrame, a Polars frame, a DuckDB connection or a record batch reader. Writes are supported too: append, overwrite with a filter, delete by predicate and upsert against the table's identifier fields, all committing a new snapshot through the catalog. It also exposes the maintenance surface most people otherwise need Spark for, including schema evolution, partition evolution, branches, tags and snapshot expiry.
For reading and administering Iceberg tables from Python without a JVM, PyIceberg is the reference implementation and there is no real competitor. Treat it as a file planner and a catalog client, not as an engine, and pair it with DuckDB or Polars for anything that needs to compute.
Use it if
- Your Iceberg tables are written by Spark, Flink or Trino and you want to read them from a Python service, a notebook or a Lambda without standing up a JVM
- You want file pruning rather than a full table read: scan(row_filter=...) uses manifest statistics so only the Parquet files that can contain matching rows get opened
- You need catalog and table administration from Python: create namespaces and tables, evolve the schema, add or drop partition fields, cut branches and tags, expire old snapshots
- You are producing modest volumes from Python, for example a job that appends a few batches per hour into a table that a real engine compacts later
- You need to hand data to something else: to_arrow, to_pandas, to_polars, to_duckdb and to_arrow_batch_reader all come out of the same scan
- You are expecting a query engine. PyIceberg selects columns and prunes files, then hands you the rows. Joins, aggregation and anything beyond the row filter happen in DuckDB, Polars or DataFusion, in your process, using your memory. A filter that matches most of a large table will not end well.
- Small frequent appends are your write pattern. PyIceberg defaults to fast append, which keeps commits cheap but produces a lot of metadata, and automatic compaction is still a planned feature tracked in an open issue rather than something that ships today.
- You want one install line. The base package cannot read data at all; you need pyiceberg[pyarrow] plus a catalog extra such as glue, hive, sql-postgres or rest-sigv4, plus a filesystem extra such as s3fs, adlfs or gcsfs. Get the combination wrong and the failure arrives as an import error or a first-scan stack trace, not as a clear message.
- You have several writers hitting one table. Commits go through the catalog and can lose a conflict, so every writer needs retry logic around the whole read-modify-commit cycle; PyIceberg does not add that for you.
- Your tables are Delta Lake, Hudi or plain Parquet in a bucket. This library only speaks the Iceberg spec, and deltalake or pyarrow.dataset is the right tool for those.
- You need timestamps below microsecond resolution. TimestampType tops out at microseconds, and nanosecond Arrow columns are rejected on write unless you opt into downcast-ns-timestamp-to-us-on-write, which silently loses precision.
Setup reality
The install you actually want looks like pip install "pyiceberg[pyarrow,s3fs,glue]", chosen from three orthogonal axes: the data reader, the catalog and the object store. Even the base package is not small, pulling mmh3, requests, click, rich, strictyaml, pydantic, fsspec, pyparsing, tenacity, pyroaring, cachetools and zstandard, and the pyarrow extra adds both pyarrow 17 or newer and the Rust pyiceberg-core wheel. Watch the pydantic constraint: 2.4.0, 2.4.1, 2.12.0 and 2.12.1 are explicitly excluded, so a service pinned to one of those will simply fail to resolve. Catalog configuration lives in a .pyiceberg.yaml searched for in PYICEBERG_HOME, then the home directory, then the working directory, or you skip the file and pass the properties straight to load_catalog. There is also an environment variable form where double underscore means a nested key and single underscore becomes a dash, so PYICEBERG_CATALOG__DEFAULT__S3__ACCESS_KEY_ID sets s3.access-key-id on the default catalog, which is the shape you want in Kubernetes. Python 3.10 or newer and below 4 is required.
Patterns
Connect to a catalog without a config fileload-catalog
from pyiceberg.catalog import load_catalog
catalog = load_catalog(
"prod",
**{
"type": "rest",
"uri": "https://rest-catalog:8181/",
"warehouse": "s3://my-warehouse/",
"s3.access-key-id": ACCESS_KEY,
"s3.secret-access-key": SECRET_KEY,
},
)
table = catalog.load_table("analytics.events")The alternative is a .pyiceberg.yaml found via PYICEBERG_HOME, the home directory or the working directory, in that order. In a container prefer the environment variable form: PYICEBERG_CATALOG__PROD__S3__ACCESS_KEY_ID maps to s3.access-key-id on the prod catalog, since double underscore nests and single underscore becomes a dash.
Create a namespace and a table from an Arrow schemacreate-table
import pyarrow as pa
catalog.create_namespace_if_not_exists("analytics")
df = pa.Table.from_pylist([
{"city": "Amsterdam", "lat": 52.371807, "long": 4.896029},
])
tbl = catalog.create_table("analytics.cities", schema=df.schema)Passing an Arrow schema saves you hand-writing NestedField definitions with explicit field ids, but it also means nullability comes straight from Arrow: build the schema with nullable=False on the fields you want required, because you cannot tighten that later without a rewrite.
Write data into the tableappend-and-overwrite
tbl.append(df)
# replace everything
tbl.overwrite(df)
# replace only the rows matching a predicate
from pyiceberg.expressions import EqualTo
tbl.overwrite(new_rows, overwrite_filter=EqualTo("city", "Paris"))Both take a pyarrow.Table, and mismatched types fail at write time, so build the frame with schema=tbl.schema().as_arrow() when the data comes from Python dicts. Each call commits its own snapshot; if you need several changes to land atomically, do them inside tbl.transaction().
Read a slice of a large tablescan-with-filter
from pyiceberg.expressions import And, GreaterThanOrEqual, EqualTo
scan = tbl.scan(
row_filter=And(GreaterThanOrEqual("trip_distance", 10.0), EqualTo("VendorID", 2)),
selected_fields=("VendorID", "tpep_pickup_datetime", "trip_distance"),
limit=1000,
)
arrow_table = scan.to_arrow()
# same thing with the string grammar
tbl.scan(row_filter="trip_distance >= 10.0 AND VendorID = 2").to_arrow()The filter prunes files using manifest statistics, then the remaining rows are filtered in memory, so a predicate on a partition or sorted column removes far more work than one on a random column. limit caps rows returned, not files read. Use plan_files() when you want to see which Parquet files a filter actually selected.
Hand the scan to pandas, Polars or DuckDBread-into-dataframes
pdf = tbl.scan(row_filter="city = 'Paris'").to_pandas()
pl_df = tbl.scan().to_polars()
con = tbl.scan(selected_fields=("city", "lat")).to_duckdb(table_name="cities")
con.execute("SELECT city FROM cities ORDER BY lat DESC LIMIT 5").fetchall()
for batch in tbl.scan().to_arrow_batch_reader():
process(batch)to_pandas, to_polars and to_duckdb all materialise the result in this process, so on anything large use to_arrow_batch_reader and stream. Each of these needs its own extra installed; calling to_duckdb without duckdb present raises at the call, not at import.
Read the table as of an older snapshottime-travel
snapshots = tbl.inspect.snapshots() # arrow table of the history
previous = snapshots.column("snapshot_id")[-2].as_py()
old = tbl.scan(snapshot_id=previous).to_arrow()
# metadata tables travel too
tbl.inspect.entries(snapshot_id=previous)inspect.snapshots() and inspect.refs() are the two metadata tables that do not accept snapshot_id, because they describe the history itself. Time travel only reaches snapshots that still exist, so anything you expired is gone regardless of what the metadata log remembers.
Merge a batch against identifier fieldsupsert-rows
result = tbl.upsert(df)
print(result.rows_updated, result.rows_inserted)
# or state the join keys explicitly
tbl.upsert(df, join_cols=["city"], when_not_matched_insert_all=False)Without join_cols it uses the table's identifier-field-ids, and a table created without them raises rather than guessing. Upsert reads the matching existing rows to work out the diff, so cost scales with how much of the table your keys touch, not with the size of the incoming batch.
Delete by predicatedelete-rows
tbl.delete(delete_filter="city == 'Paris'")
from pyiceberg.expressions import LessThan
tbl.delete(delete_filter=LessThan("event_ts", "2026-01-01T00:00:00"))When every row in a Parquet file matches, the file is dropped from the manifest without being opened, which is fast. When only some rows match, that file is read and rewritten, so a predicate that scatters across many files turns a delete into a full rewrite of them.
Add, rename and drop columns safelyschema-evolution
from pyiceberg.types import IntegerType, StringType
with tbl.update_schema() as update:
update.add_column("retries", IntegerType(), "Number of retries")
update.rename_column("usr", "user")
update.add_column(("details", "confirmed_by"), StringType())
with tbl.update_schema() as update:
update.union_by_name(incoming_schema)Field ids are assigned for you, which is what makes a rename a metadata-only change instead of a rewrite. Anything the spec calls breaking is refused unless you pass allow_incompatible_changes=True to update_schema, and nested fields are addressed as tuples rather than dotted strings.
Change how the table is partitionedpartition-evolution
from pyiceberg.transforms import BucketTransform, DayTransform
with tbl.update_spec() as update:
update.add_field("event_ts", DayTransform(), "day_ts")
update.add_field("user_id", BucketTransform(16), "bucket_user")
update.remove_field("old_partition_name")Iceberg keeps the old spec for existing data rather than rewriting it, so after this change the table holds files under two layouts and readers handle both. remove_field takes the partition field name, not the source column name, which is why naming the field in add_field is worth doing.
Pin a snapshot or work on a branchbranches-and-tags
snapshot_id = tbl.current_snapshot().snapshot_id
tbl.manage_snapshots().create_tag(snapshot_id, "release-2026-08").commit()
with tbl.manage_snapshots() as ms:
ms.create_branch(snapshot_id, "backfill")
tbl.append(df, branch="backfill")Nothing happens until commit() is called, or until the context manager exits, which is the usual reason a tag appears not to have been created. Tags are immutable pointers meant for retention; branches are mutable and every write method takes a branch argument to target one.
Clean up old metadataexpire-snapshots
from datetime import datetime, timedelta
tbl.maintenance.expire_snapshots().older_than(
datetime.now() - timedelta(days=7)
).commit()
with tbl.maintenance.expire_snapshots() as expire:
expire.by_id(805611270568163028)This is destructive: expired snapshots stop being reachable for time travel and any tag or branch pointing at them becomes the only thing keeping them alive. It also does not compact anything, so a table fed by many small appends still accumulates data files that only a real engine can rewrite today.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| deltalake | PyPI | Your lakehouse is Delta Lake instead of Iceberg and you want the same read and write shape from Python |
| duckdb | PyPI | You mostly want SQL over the data and its Iceberg extension covers the reads you need |
| daft | PyPI | The dataset is larger than one machine's memory and you want a distributed dataframe that reads Iceberg directly |