Have you ever spent countless hours meticulously building a data ingestion pipeline, only to have a subtle, unannounced schema change in an upstream API completely break your downstream analytics? It’s a familiar sting for anyone dealing with dynamic external feeds. While we’ve previously tackled the initial challenge of architecting resilient, schema-aware event ingestion, getting that data *into* your lake is only half the battle. The true nightmare begins when you need to maintain data quality, evolve schemas gracefully, perform atomic updates, or even roll back to a previous state—all while avoiding the complexity and cost of a traditional data warehouse. I recently navigated this exact problem with a critical content feed, where schema drift was a constant threat. This post shares how I leveraged Apache Iceberg, specifically with PyIceberg, to transform a chaotic data lake into a reliable, versioned data lakehouse, enabling database-like guarantees right on top of object storage. You'll learn how to set up Iceberg locally, ingest and manage dynamic data, and lay the groundwork for schema evolution, empowering you to build truly robust data platforms.
Key Takeaways
- Apache Iceberg provides a table format that brings ACID transactions and schema evolution capabilities directly to your data lake, moving beyond simple file storage.
- PyIceberg offers a Pythonic interface to manage Iceberg tables, allowing for programmatic schema definition, data ingestion, and table maintenance.
- Schema evolution in Iceberg (e.g., adding columns, safely changing types) is managed through metadata, ensuring backward compatibility and preventing data corruption.
- Time travel capabilities allow querying historical states of the table, crucial for auditing, reproducibility, and error recovery.
- Setting up a local Iceberg environment with a file-based catalog is an excellent way to prototype and understand its core features before scaling to distributed systems.
The Problem: Beyond Simple File Ingestion for Dynamic Feeds
My previous work focused on how to robustly ingest dynamic, semi-structured data from external feeds, ensuring schema awareness and resilience against malformed inputs. But what happens after ingestion? Most data lakes end up as vast collections of Parquet or ORC files. While efficient for storage, this structure lacks crucial database features: atomic writes, schema enforcement, schema evolution, and the ability to "time travel" to past states. When an external feed inevitably changes its structure, adding a new field or changing a data type, you're faced with a painful choice: rewrite historical data, or build complex ETL processes to handle multiple schema versions, leading to inconsistent analytics and brittle pipelines. I needed a solution that would give me the flexibility of a data lake with the reliability of a data warehouse, especially for critical, ever-changing content.
Data and Sources
For this demonstration, I'm using the Shopify Engineering blog's RSS feed, which provides a real-world example of dynamic, semi-structured content. It's an excellent source because blog posts are often updated, and the feed itself can subtly change its structure over time. We'll parse this XML-based feed into a structured format for ingestion.
- Shopify Engineering Blog RSS Feed: https://shopify.engineering/blogs/engineering.atom
- PyIceberg Documentation: https://py.iceberg.apache.org/main/
- Apache Iceberg Project: https://iceberg.apache.org/
Data accessed on 2026-10-08.
Step 1 — Re-Ingesting Dynamic Feed Data into a Staging DataFrame
The first sub-problem is to take our raw, semi-structured feed data and transform it into a structured format that Iceberg can easily consume. Building on our prior ingestion knowledge, I'll use `feedparser` to fetch and parse the RSS feed, then normalize the entries into a Pandas DataFrame. This step ensures that the data is clean and consistently formatted before it hits our lakehouse.
import feedparser
import pandas as pd
import requests
def fetch_and_stage_feed_data(feed_url: str) -> pd.DataFrame:
"""Fetches RSS feed and stages relevant data into a Pandas DataFrame."""
try:
response = requests.get(feed_url, timeout=10)
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
feed = feedparser.parse(response.content)
except requests.exceptions.RequestException as e:
print(f"Error fetching feed: {e}")
return pd.DataFrame() # Return empty DataFrame on failure
except Exception as e:
print(f"Error parsing feed: {e}")
return pd.DataFrame()
entries_data = []
for entry in feed.entries:
# Extract and normalize fields, handling missing ones
entries_data.append({
'id': entry.id,
'title': entry.title,
'link': entry.link,
'published': entry.published,
'summary': getattr(entry, 'summary', ''), # Use getattr for optional fields
'author': getattr(entry, 'author', 'Unknown')
})
return pd.DataFrame(entries_data)
This code snippet defines a function that fetches the RSS feed, handles potential network errors, and then iterates through each entry. It extracts key fields like `id`, `title`, `link`, `published`, `summary`, and `author`, using `getattr` to gracefully handle cases where a field might be missing in some entries. The result is a clean Pandas DataFrame, which acts as our staging area before Iceberg ingestion.
Step 2 — Setting Up the Local Iceberg Lakehouse Environment
Before we can write data to an Iceberg table, we need an Iceberg catalog and a storage location. For local development and demonstration, a file-based catalog is ideal. It allows us to interact with Iceberg tables directly on our local filesystem without needing a distributed cluster like Spark or Flink. This step involves initializing a `pyiceberg` catalog that points to a local directory, which will store all the Iceberg metadata and data files.
from pyiceberg.catalog import Catalog, load_catalog
import os
def setup_local_iceberg_environment(warehouse_path: str) -> Catalog:
"""Sets up a local file-based Iceberg catalog."""
if not os.path.exists(warehouse_path):
os.makedirs(warehouse_path)
print(f"Created Iceberg warehouse directory: {warehouse_path}")
# Configuration for a local file-based catalog
# The 'uri' points to the base directory where table metadata will be stored
catalog_conf = {
"type": "rest", # Using rest for local file-based catalog
"uri": f"file://{warehouse_path}",
"warehouse": warehouse_path # This is where data files will be stored
}
# Load the catalog with the specified configuration
catalog = load_catalog("local_file_catalog", catalog_conf)
print(f"Initialized local Iceberg catalog at: {warehouse_path}")
return catalog
The `setup_local_iceberg_environment` function ensures our specified `warehouse_path` exists. It then configures and loads a `rest` type catalog, which for local file systems, essentially uses the file path as its URI and warehouse location. This catalog object is our entry point for creating, loading, and managing Iceberg tables.
Step 3 — Creating and Populating the Initial Iceberg Table
With our data staged and our Iceberg environment ready, the next sub-problem is to define the table's initial schema and write our first batch of data. This is where we establish the foundation for all future operations, including schema evolution and time travel. We'll infer a basic schema from our DataFrame and then use `pyiceberg` to create the table and append the data.
from pyiceberg.schema import Schema
from pyiceberg.types import NestedField, LongType, StringType, TimestampType, StructType
from pyiceberg.io.pyarrow import PyArrowFileIO
import pyarrow as pa
import pyarrow.parquet as pq
def create_and_populate_iceberg_table(
catalog: Catalog,
table_name: str,
df: pd.DataFrame,
warehouse_path: str
):
"""Creates an Iceberg table and populates it with initial data."""
# Convert DataFrame to PyArrow Table
arrow_table = pa.Table.from_pandas(df)
# Define Iceberg schema from PyArrow Table schema
# Iceberg schemas are explicit and enforce types
iceberg_schema = Schema(
NestedField(1, "id", StringType(), required=True),
NestedField(2, "title", StringType(), required=True),
NestedField(3, "link", StringType(), required=True),
NestedField(4, "published", TimestampType(), required=True),
NestedField(5, "summary", StringType(), required=False),
NestedField(6, "author", StringType(), required=False)
)
try:
# Create the Iceberg table
table = catalog.create_table(
identifier=table_name,
schema=iceberg_schema,
properties={"format-version": "2"}, # Use format version 2 for advanced features
location=os.path.join(
