AWS Builder Center

Builder Center

Exploring Amazon S3 Tables with PyIceberg

Learn how to query and manage Apache Iceberg tables on Amazon S3 Tables directly from a local Python script using PyIceberg, no Spark, Glue, or EMR cluster required.

Steven Cook

If you've worked with Apache Iceberg on AWS, you probably associate it with Spark jobs, Glue crawlers, and EMR clusters. That's a lot of infrastructure just to poke around a table. But what if you could connect to your Iceberg tables on  Amazon S3 Tables  from a plain Python script on your laptop — no Spark, no Glue, no cluster at all?
That's exactly what  PyIceberg  lets you do.

What Are S3 Tables?

Amazon S3 Tables is a purpose-built S3 storage type optimized for tabular data in Apache Iceberg format. Instead of managing your own Iceberg catalog and metastore, S3 Tables provides a built-in Iceberg REST catalog endpoint. You point your tooling at it, authenticate with SigV4, and you're working with Iceberg tables backed by S3 — with automatic compaction and snapshot management handled for you.
The REST endpoint lives at:
https://s3tables.<REGION>.amazonaws.com/iceberg
Any Iceberg-compatible client that speaks the REST catalog protocol can connect. That includes Spark, Trino, and PyIceberg.

What Is PyIceberg?

PyIceberg is the official Python implementation of the Apache Iceberg spec. It's catalog-centric: you load a catalog, then interact with namespaces and tables through it. Under the hood it reads and writes Parquet files via PyArrow, which means you get a fast, native Python data path without any JVM overhead.
Key capabilities from the  PyIceberg API docs :
  • Load catalogs (REST, Hive, Glue, etc.) with  load_catalog() 
  • Create, list, load, rename, and drop tables
  • Read data into PyArrow Tables, Pandas DataFrames, DuckDB, Polars, or Ray datasets
  • Write data with  append() ,  overwrite() , and  delete() 
  • Inspect metadata: snapshots, partitions, manifests, data files
  • Schema evolution, partition evolution, and sort order updates
  • Time travel via snapshot IDs
The important thing: all of this runs in a single Python process. No executors, no driver, no cluster manager.

Connecting to S3 Tables from PyIceberg

The connection setup is minimal. Here's the catalog configuration from our demo:
from pyiceberg.catalog import load_catalog

catalog = load_catalog(
    "s3tables_catalog",
    **{
        "type": "rest",
        "uri": f"https://s3tables.{region}.amazonaws.com/iceberg",
        "warehouse": table_bucket_arn,
        "rest.sigv4-enabled": "true",
        "rest.signing-name": "s3tables",
        "rest.signing-region": region,
    },
)
That's it. No Spark config, no Glue connection, no Hadoop classpath. PyIceberg uses your existing AWS credentials (environment variables, profile, or IAM role) and signs requests with SigV4 automatically.

A Walkthrough: Orders Table on S3 Tables

Let's walk through a realistic workflow. We'll create a namespace, define a table, insert data, query it, update rows, and inspect the snapshot history — all from a local Python script.

1. Create a Namespace and Table

S3 Tables supports single-level namespaces. Creating one and defining a table is straightforward:
import pyarrow as pa

catalog.create_namespace("demo_ns")

schema = pa.schema([
    pa.field("order_id", pa.int32()),
    pa.field("customer_name", pa.string()),
    pa.field("product", pa.string()),
    pa.field("quantity", pa.int32()),
    pa.field("price", pa.float64()),
    pa.field("status", pa.string()),
])

table = catalog.create_table("demo_ns.orders", schema=schema)
Notice we're using a PyArrow schema directly. PyIceberg handles the conversion to Iceberg types internally.

2. Insert Data

Writing data uses PyArrow tables and the  append()  method:
rows = [
    {"order_id": 1, "customer_name": "Alice",   "product": "Widget A",
     "quantity": 10, "price": 29.99, "status": "shipped"},
    {"order_id": 2, "customer_name": "Bob",     "product": "Widget B",
     "quantity": 5,  "price": 49.99, "status": "pending"},
    {"order_id": 3, "customer_name": "Charlie", "product": "Widget C",
     "quantity": 2,  "price": 99.99, "status": "shipped"},
]

arrow_table = pa.Table.from_pylist(rows, schema=schema)
table.append(arrow_table)
Each  append()  call creates a new Iceberg snapshot with new Parquet data files. PyIceberg uses fast append by default, which minimizes the amount of data written per commit.

3. Query Data

Scanning the table returns data as a Pandas DataFrame (or Arrow table, DuckDB cursor, Polars frame):
# Full scan
df = table.scan().to_pandas()

# Filtered scan with predicate pushdown
from pyiceberg.expressions import EqualTo

df_shipped = table.scan(
    row_filter=EqualTo("status", "shipped")
).to_pandas()
The filter is pushed down to the Iceberg metadata layer. PyIceberg reads partition stats and column-level min/max bounds to skip Parquet files that can't contain matching rows. You get the performance benefit of Iceberg's metadata without writing a single line of Spark.
You can also stream results to avoid loading everything into memory:
for batch in table.scan().to_arrow_batch_reader():
    process(batch)

4. Update Rows (Read-Transform-Overwrite)

PyIceberg doesn't have a SQL-style  UPDATE  statement. The pattern is read the data, transform it in memory, and overwrite:
df = table.scan().to_pandas()
df.loc[df["status"] == "pending", "status"] = "processing"
table.overwrite(pa.Table.from_pandas(df, schema=schema))
This rewrites the entire table, which is fine for small-to-medium datasets. For tables with millions of rows where you're updating a small subset, the Spark/Glue approach with row-level updates is more efficient — but for exploration, prototyping, and moderate workloads, this pattern works well.

5. Time Travel

Every write creates a snapshot. You can query any previous version of the table:
snapshots = table.metadata.snapshots

# Read the table as it was at the first snapshot
old_snapshot = snapshots[0]
df_old = table.scan(snapshot_id=old_snapshot.snapshot_id).to_pandas()
This is useful for auditing, debugging data issues, or comparing table states before and after a pipeline run.

6. Inspect Metadata

PyIceberg exposes rich metadata inspection through the  table.inspect  API:
# Snapshot history
table.inspect.snapshots()

# Partition information
table.inspect.partitions()

# Manifest files
table.inspect.manifests()

# Individual data files with readable metrics
table.inspect.files()

# Metadata log entries
table.inspect.metadata_log_entries()
Each of these returns a PyArrow table, so you can filter and analyze metadata the same way you'd analyze data. Want to know how many files were added in the last snapshot? Which partitions have the most records? It's all there without spinning up a Spark session.

When to Use PyIceberg vs. Spark

This isn't an either/or decision. They serve different use cases:
Use Case
PyIceberg
Spark (Glue/EMR)
Exploring table structure and metadata
✅ Great
Overkill
Ad-hoc queries on small-to-medium data
✅ Great
Works but slow to start
Prototyping data pipelines
✅ Great
Heavy setup
CI/CD data validation checks
✅ Great
Slower startup
Row-level UPDATE / DELETE / MERGE
❌ Not supported
✅ Native support
Processing billions of rows
❌ Single process
✅ Distributed
Production ETL at scale
Depends on volume
✅ Built for this
The sweet spot for PyIceberg is anywhere you'd normally fire up a notebook or write a quick script. Schema checks before a deploy, data quality assertions in a CI pipeline, quick exploration of what a table looks like after a pipeline ran, all of these are faster and cheaper without a cluster.

Getting Started

The setup is minimal:
pip install "pyiceberg[pandas,pyarrow,s3fs]" boto3
Then run the demo script against your S3 table bucket:
"""
Demo: Using PyIceberg with Amazon S3 Tables (no Spark/Glue required).

Connects directly to the S3 Tables Iceberg REST endpoint to create a table,
insert rows, query data, update rows, and inspect snapshots.

Prerequisites:
  - An S3 table bucket already created in your AWS account
  - AWS credentials configured (via env vars, profile, or IAM role)
  - pip install -r requirements.txt

Reference:
  https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-open-source.html

Usage:
  python s3tables_demo.py \
    --table-bucket-arn arn:aws:s3tables:us-east-1:123456789012:bucket/my-bucket \
    --namespace my_namespace \
    --table orders
"""

import argparse
import pyarrow as pa
import pandas as pd
from pyiceberg.catalog import load_catalog
from pyiceberg.expressions import EqualTo, GreaterThanOrEqual

# ---------------------------------------------------------------------------
# Catalog setup
# ---------------------------------------------------------------------------

def get_catalog(table_bucket_arn: str, region: str):
    """
    Connect to the S3 Tables Iceberg REST endpoint using SigV4 auth.

    The key config properties:
      - uri:  https://s3tables.<REGION>.amazonaws.com/iceberg
      - warehouse: your table bucket ARN
      - rest.sigv4-enabled / rest.signing-name: for SigV4 authentication
    """
    return load_catalog(
        "s3tables_catalog",
        **{
            "type": "rest",
            "uri": f"https://s3tables.{region}.amazonaws.com/iceberg",
            "warehouse": table_bucket_arn,
            "rest.sigv4-enabled": "true",
            "rest.signing-name": "s3tables",
            "rest.signing-region": region,
        },
    )

# ---------------------------------------------------------------------------
# Schema & sample data
# ---------------------------------------------------------------------------

ORDER_SCHEMA = pa.schema([
    pa.field("order_id", pa.int32()),
    pa.field("customer_name", pa.string()),
    pa.field("product", pa.string()),
    pa.field("quantity", pa.int32()),
    pa.field("price", pa.float64()),
    pa.field("status", pa.string()),
])

SAMPLE_ORDERS = [
    {"order_id": 1, "customer_name": "Alice",   "product": "Widget A", "quantity": 10, "price": 29.99, "status": "shipped"},
    {"order_id": 2, "customer_name": "Bob",     "product": "Widget B", "quantity": 5,  "price": 49.99, "status": "pending"},
    {"order_id": 3, "customer_name": "Charlie", "product": "Widget C", "quantity": 2,  "price": 99.99, "status": "shipped"},
    {"order_id": 4, "customer_name": "Diana",   "product": "Widget A", "quantity": 7,  "price": 29.99, "status": "delivered"},
    {"order_id": 5, "customer_name": "Eve",     "product": "Widget B", "quantity": 3,  "price": 49.99, "status": "pending"},
]

# ---------------------------------------------------------------------------
# Demo steps
# ---------------------------------------------------------------------------

def create_namespace(catalog, namespace: str):
    """Create a namespace (S3 Tables only supports single-level namespaces)."""
    existing = [ns[0] for ns in catalog.list_namespaces()]
    if namespace in existing:
        print(f"  Namespace '{namespace}' already exists, skipping creation.")
        return
    catalog.create_namespace(namespace)
    print(f"  Created namespace '{namespace}'.")

def create_table(catalog, namespace: str, table_name: str):
    """Create an Iceberg table from our schema."""
    identifier = f"{namespace}.{table_name}"
    tables = [t[1] for t in catalog.list_tables(namespace)]
    if table_name in tables:
        print(f"  Table '{identifier}' already exists, skipping creation.")
        return catalog.load_table(identifier)

    table = catalog.create_table(identifier, schema=ORDER_SCHEMA)
    print(f"  Created table '{identifier}'.")
    return table

def insert_rows(table, rows: list[dict]):
    """Append rows to the table using PyArrow."""
    arrow_table = pa.Table.from_pylist(rows, schema=ORDER_SCHEMA)
    table.append(arrow_table)
    print(f"  Inserted {len(rows)} rows.")

def query_all(table) -> pd.DataFrame:
    """Full table scan → pandas DataFrame."""
    df = table.scan().to_pandas()
    print(f"\n  All rows ({len(df)}):")
    print(df.to_string(index=False))
    return df

def query_with_filter(table, column: str, value):
    """Filter scan using an Iceberg expression."""
    df = table.scan(row_filter=EqualTo(column, value)).to_pandas()
    print(f"\n  Rows where {column} == '{value}' ({len(df)}):")
    print(df.to_string(index=False))
    return df

def update_status(table, from_status: str, to_status: str):
    """
    Read matching rows, update in-memory, overwrite the table.
    PyIceberg doesn't have row-level UPDATE — the pattern is
    read → transform → overwrite.
    """
    df = table.scan().to_pandas()
    mask = df["status"] == from_status
    count = mask.sum()
    df.loc[mask, "status"] = to_status
    table.overwrite(pa.Table.from_pandas(df, schema=ORDER_SCHEMA))
    print(f"  Updated {count} rows: status '{from_status}' → '{to_status}'.")

def show_snapshots(table):
    """Print the snapshot history (time-travel metadata)."""
    snapshots = table.metadata.snapshots
    print(f"\n  Snapshot history ({len(snapshots)} snapshots):")
    for snap in snapshots:
        print(f"    id={snap.snapshot_id}  ts={snap.timestamp_ms}  summary={snap.summary}")

def time_travel_query(table):
    """
    Query a previous version of the table using Iceberg time travel.

    Uses the first (oldest) snapshot to read the data as it existed
    before any updates were applied.
    """
    snapshots = table.metadata.snapshots
    if len(snapshots) < 2:
        print("  Not enough snapshots for time travel (need at least 2).")
        return None

    old_snapshot = snapshots[0]
    print(f"  Querying snapshot {old_snapshot.snapshot_id} "
          f"(ts={old_snapshot.timestamp_ms})...")

    df = table.scan(snapshot_id=old_snapshot.snapshot_id).to_pandas()
    print(f"\n  Rows at oldest snapshot ({len(df)}):")
    print(df.to_string(index=False))
    return df

# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def parse_args():
    p = argparse.ArgumentParser(description="PyIceberg + S3 Tables demo (no Spark)")
    p.add_argument("--table-bucket-arn", required=True,
                   help="ARN of your S3 table bucket, e.g. arn:aws:s3tables:us-east-1:123456789012:bucket/my-bucket")
    p.add_argument("--namespace", default="demo_ns",
                   help="Namespace (single-level) to use (default: demo_ns)")
    p.add_argument("--table", default="orders",
                   help="Table name (default: orders)")
    p.add_argument("--region", default=None,
                   help="AWS region (auto-detected from ARN if omitted)")
    return p.parse_args()

def region_from_arn(arn: str) -> str:
    # arn:aws:s3tables:<region>:<account>:bucket/<name>
    parts = arn.split(":")
    return parts[3]

def main():
    args = parse_args()
    region = args.region or region_from_arn(args.table_bucket_arn)

    print(f"Region:    {region}")
    print(f"Bucket:    {args.table_bucket_arn}")
    print(f"Namespace: {args.namespace}")
    print(f"Table:     {args.table}\n")

    # 1. Connect to S3 Tables REST catalog
    print("1) Connecting to S3 Tables Iceberg REST endpoint...")
    catalog = get_catalog(args.table_bucket_arn, region)
    print("   Connected.\n")

    # 2. Create namespace
    print("2) Creating namespace...")
    create_namespace(catalog, args.namespace)

    # 3. Create table
    print("\n3) Creating table...")
    table = create_table(catalog, args.namespace, args.table)

    # 4. Insert sample rows
    print("\n4) Inserting sample data...")
    insert_rows(table, SAMPLE_ORDERS)

    # 5. Query all rows
    print("\n5) Querying all rows...")
    query_all(table)

    # 6. Filtered query
    print("\n6) Filtered query (status == 'shipped')...")
    query_with_filter(table, "status", "shipped")

    # 7. Update rows
    print("\n7) Updating pending → processing...")
    update_status(table, "pending", "processing")

    # 8. Verify update
    print("\n8) Verifying update...")
    query_with_filter(table, "status", "processing")

    # 9. Snapshot history
    print("\n9) Snapshot history (time travel)...")
    show_snapshots(table)

    # 10. Time travel query — read data from before the update
    print("\n10) Time travel query (reading oldest snapshot)...")
    time_travel_query(table)

    print("\nDone.")

if __name__ == "__main__":
    main()
You'll need an S3 table bucket (create one in the S3 console) and IAM permissions for the  s3tables:*  actions on that bucket.

Wrapping Up

S3 Tables gives you managed Iceberg storage with a standard REST catalog interface. PyIceberg gives you a lightweight Python client that speaks that interface natively. Together, they let you work with production Iceberg tables from a local Python environment, no Spark cluster, no Glue job, no JVM.
For exploration, prototyping, and lightweight operations, that's a significant reduction in complexity and cost. Save the Spark cluster for when you actually need distributed compute.

Resources

# amazon-s3# tables
Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.
Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article