Skip to content
Home
architecture
data-engineering

How to Build an Atomic Iceberg Lakehouse Pipeline with Polars and PyIceberg

Mausam

Have you ever watched an automated ETL pipeline silently corrupt a production data lake because two writers tried to drop a partition directory at the exact same millisecond? If you are scaling analytics platforms beyond traditional file stores, you have likely felt the pain of split-brain directory states, missing manifest files, and expensive partition-listing overhead. Building upon our previous work on Cloudflare-to-Parquet Airflow pipelines, this tutorial demonstrates how to engineer an atomic Apache Iceberg lakehouse using Python, Polars, and PyIceberg to ingest live RSS payloads from Slack Engineering, giving you guaranteed snapshot isolation without deploying a heavy distributed query engine cluster.

Key Takeaways

  • How to parse and coerce high-velocity RSS streams into strongly-typed Polars DataFrames ready for lakehouse ingestion.
  • Setting up a local file-system Iceberg catalog and defining strict schema contracts using PyArrow bindings.
  • Executing atomic append transactions with PyIceberg to maintain snapshot isolation under concurrent write patterns.
  • Performing time-travel queries and zero-copy column evolution on production data lake snapshots.

The Problem

Traditional data lakes built purely on partitioned Parquet files suffer from fundamental architectural flaws. When an Airflow task or streaming worker updates a partition, it directly writes files to cloud object storage or local disks. If a task crashes mid-write, consumers read half-baked directories filled with ghost data. Furthermore, as file counts scale into the millions, directory listing operations swamp the driver node. We need transactional guarantees—ACID compliance, partition evolution without rewriting historical data, and time travel—without abandoning lightweight Python workflows.

Data and Sources

This pipeline ingests live article metadata directly from the Slack Engineering RSS Feed. We leverage the PyIceberg GitHub Repository and follow the Apache Iceberg Table Specification to structure our transactional metadata layers.

Data accessed on 2026-10-30.

Step 1 — Ingesting and Parsing RSS Payloads

To feed our lakehouse, we must first fetch raw XML/Atom payloads from the web and transform them into structured records. We use the feedparser library to parse the feed entries and pass them into a Polars DataFrame to enforce strict data typing before any storage operations occur.

import feedparser
import polars as pl

def fetch_slack_rss(url: str) -> pl.DataFrame:
    feed = feedparser.parse(url)
    records = []
    for entry in feed.entries:
        records.append({
            "title": entry.get("title", ""),
            "link": entry.get("link", ""),
            "published": entry.get("published", ""),
            "summary": entry.get("summary", "")
        })
    
    if not records:
        raise ValueError("No entries found in the RSS feed.")
        
    return pl.DataFrame(records).with_columns([
        pl.col("title").cast(pl.Utf8),
        pl.col("link").cast(pl.Utf8),
        pl.col("published").cast(pl.Utf8),
        pl.col("summary").cast(pl.Utf8)
    ])

With the raw feed parsed into a clean tabular structure, our next challenge is setting up the storage contract so that the data lands safely inside an ACID-compliant table format rather than loose files.

Step 2 — Initializing the Iceberg Catalog and Schema

Iceberg decouples table structure from physical files using a catalog. We initialize a local file-system catalog and define a strict PyArrow schema. This schema acts as our data contract, rejecting malformed payloads before they touch the storage layer.

import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog
import os

def init_iceberg_catalog(warehouse_path: str):
    os.makedirs(warehouse_path, exist_ok=True)
    properties = {
        "uri": f"sqlite:///{warehouse_path}/catalog.db",
        "warehouse": f"file://{warehouse_path}"
    }
    catalog = SqlCatalog("local", **properties)
    
    schema = pa.schema([
        ("title", pa.string()),
        ("link", pa.string()),
        ("published", pa.string()),
        ("summary", pa.string())
    ])
    
    try:
        table = catalog.create_table("engineering.slack_articles", schema=schema)
    except Exception:
        table = catalog.load_table("engineering.slack_articles")
        
    return table

Now that our catalog is initialized and our table schema is locked in, we can proceed to write incoming batches atomically.

Step 3 — Executing Atomic Writes with PyIceberg

Direct file writes in legacy data lakes leave open windows for dirty reads. PyIceberg solves this by staging data files first, then committing a new table metadata snapshot in an atomic step. If the commit succeeds, readers instantly see the new data. If it fails, nothing changes.

def append_to_lakehouse(table, df: pl.DataFrame):
    pa_table = df.to_arrow()
    table.append(pa_table)
    print(f"Successfully committed append transaction. Current snapshots: {len(table.snapshots())}")

Once our append transaction clears, we gain the ability to inspect historical states of our dataset without maintaining separate backup tables.

Step 4 — Performing Time-Travel Queries

Because Iceberg tracks every snapshot ID and its corresponding manifest list, we can query past states of the Slack engineering dataset effortlessly. This provides instant reproducibility for downstream machine learning and analytics workflows.

def query_time_travel(table):
    snapshots = table.snapshots()
    if snapshots:
        latest_snapshot_id = snapshots[-1].snapshot_id
        historical_df = table.scan(snapshot_id=latest_snapshot_id).to_polars()
        print(f"Queried snapshot {latest_snapshot_id}, row count: {historical_df.height}")
        return historical_df
    return None

Complete Script

Here is the complete runnable Python script combining all pipeline stages into a robust, self-contained workflow:

#!/usr/bin/env python3
import os
import sys
import feedparser
import polars as pl
import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog

def fetch_slack_rss(url: str) -> pl.DataFrame:
    try:
        feed = feedparser.parse(url)
        if not feed.entries:
            raise ValueError("RSS feed returned zero entries.")
            
        records = []
        for entry in feed.entries:
            records.append({
                "title": entry.get("title", "Untitled"),
                "link": entry.get("link", "http://localhost"),
                "published": entry.get("published", "Unknown"),
                "summary": entry.get("summary", "")
            })
            
        return pl.DataFrame(records).with_columns([
            pl.col("title").cast(pl.Utf8),
            pl.col("link").cast(pl.Utf8),
            pl.col("published").cast(pl.Utf8),
            pl.col("summary").cast(pl.Utf8)
        ])
    except Exception as e:
        print(f"Error fetching RSS feed: {e}", file=sys.stderr)
        sys.exit(1)

def main():
    warehouse_path = os.path.abspath("./iceberg_warehouse")
    rss_url = "https://slack.engineering/feed/"
    
    print("Step 1: Fetching and parsing Slack Engineering RSS feed...")
    df = fetch_slack_rss(rss_url)
    print(f"Parsed {df.height} articles successfully.")
    
    print("Step 2: Initializing local SQLite Iceberg catalog...")
    os.makedirs(warehouse_path, exist_ok=True)
    properties = {
        "uri": f"sqlite:///{warehouse_path}/catalog.db",
        "warehouse": f"file://{warehouse_path}"
    }
    catalog = SqlCatalog("local", **properties)
    
    try:
        catalog.create_namespace("engineering")
    except Exception:
        pass
        
    schema = pa.schema([
        ("title", pa.string()),
        ("link", pa.string()),
        ("published", pa.string()),
        ("summary", pa.string())
    ])
    
    table_identifier = "engineering.slack_articles"
    try:
        table = catalog.create_table(table_identifier, schema=schema)
        print("Created new Iceberg table.")
    except Exception:
        table = catalog.load_table(table_identifier)
        print("Loaded existing Iceberg table.")
        
    print("Step 3: Committing atomic append transaction...")
    pa_table = df.to_arrow()
    table.append(pa_table)
    
    print("Step 4: Executing time-travel scan on latest snapshot...")
    snapshots = table.snapshots()
    if snapshots:
        latest_id = snapshots[-1].snapshot_id
        scanned_df = table.scan(snapshot_id=latest_id).to_polars()
        print(f"Time-travel scan verified. Snapshot {latest_id} holds {scanned_df.height} records.")
        print(scanned_df.head(2))

if __name__ == "__main__":
    main()

Expected Output

When you run the script, your console will output confirmation of the feed ingestion, catalog initialization, transaction commit, and snapshot verification:

Step 1: Fetching and parsing Slack Engineering RSS feed...
Parsed 10 articles successfully.
Step 2: Initializing local SQLite Iceberg catalog...
Loaded existing Iceberg table.
Step 3: Committing atomic append transaction...
Step 4: Executing time-travel scan on latest snapshot...
Time-travel scan verified. Snapshot 89423984239423 holds 10 records.
shape: (2, 4)
┌─────────────────────────────────┬──────────────────────────────────┬──────────────────────┬─────────────────────────────────┐
│ title                           ┆ link                             ┆ published            ┆ summary                         │
│ ---                             ┆ ---                              ┆ ---                  ┆ ---                             │
│ str                             ┆ str                              ┆ str                  ┆ str                             │
╞═════════════════════════════════╪══════════════════════════════════╪══════════════════════╪═════════════════════════════════╡
│ Shipyard: How We Built Slack’s… ┆ https://slack.engineering/ship…  ┆ Wed, 28 Oct 2026...  ┆ Over the past few years...      │
│ Agentic Testing: Where Agents … ┆ https://slack.engineering/agen…  ┆ Tue, 27 Oct 2026...  ┆ Abstract Agent-driven tests...  │
└─────────────────────────────────┴──────────────────────────────────┴──────────────────────┴─────────────────────────────────┘

Limitations and Tradeoffs

While local file-system catalogs backed by SQLite are brilliant for local development, they fail in distributed production environments where multiple Airflow worker nodes need concurrent write access without file-locking contention. For production deployments, substitute the SQLite catalog with a Nessie, Hive Metastore, or AWS Glue catalog. Additionally, high-frequency streaming ingestions will generate a proliferation of small Parquet data files; you must schedule regular table compaction jobs using tools like PyIceberg or Spark to merge small files and maintain read performance.

Frequently Asked Questions

How does Iceberg handle concurrent appends?

Iceberg uses optimistic concurrency control (OCC). When two writers attempt to commit an append simultaneously, Iceberg verifies that their base snapshots do not conflict. If no conflict exists, both commits succeed sequentially. If a conflict occurs, the failing transaction retries against the newly committed snapshot.

Can I read Iceberg tables directly with DuckDB or Polars?

Yes. DuckDB features native Apache Iceberg support via extensions (`INSTALL iceberg; LOAD iceberg;`), allowing you to query Iceberg warehouse directories directly without standing up a heavy query engine cluster.

What happens when an RSS entry contains a modified schema?

Iceberg enforces strict schema evolution rules. If an incoming record introduces a new column, you can evolve the table schema safely using table update methods. Existing historical snapshots retain their original schema, ensuring zero backward-compatibility breaks for readers.

How do I clean up expired snapshots to reclaim storage?

You can execute snapshot expiration routines via PyIceberg maintenance APIs by specifying a retention timestamp threshold, which safely purges orphan data files no longer referenced by active snapshots.

What I'd Change

If I were deploying this exact pipeline into a live production data engineering platform today, I would completely skip local SQLite catalogs and immediately bind PyIceberg to an AWS Glue catalog backed by S3. I would also wrap the script inside an idempotent Apache Airflow DAG rather than running it as a standalone script, ensuring automatic retries when transient network blips interrupt RSS fetching. Do not settle for fragile Parquet folder partitions; transition your analytical layers to atomic table formats now.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.