Have you ever felt like you're playing whack-a-mole with your data pipeline, constantly patching for late-arriving records or unexpected schema changes from an external API? I’ve been there, staring at a data lake full of 'append-only' files, knowing that a critical update meant a convoluted, full-table rewrite or a messy, error-prone merge. It's a common trap: you build a robust ingestion pipeline, maybe even with the data quality checks we discussed in our previous post on Great Expectations, but what happens when the data in the storage needs to change? For data engineers dealing with the volatile reality of dynamic API sources, simply appending data isn't enough. We need a storage layer that embraces mutability without sacrificing auditability or performance. I recently tackled this challenge head-on while building a system to track evolving content from various engineering blogs, and I'm going to walk you through how Apache Iceberg became the cornerstone of my solution, enabling efficient, idempotent upserts and powerful time travel.
Key Takeaways
- Apache Iceberg's `MERGE INTO` statement provides a robust, idempotent way to handle inserts, updates, and deletes for dynamic API data, simplifying complex ETL logic.
- Time travel with Iceberg allows you to query past states of your data, crucial for auditing, debugging, and reproducing analyses without needing to manually manage historical snapshots.
- Iceberg's schema evolution capabilities mean you can adapt to changes in API responses (e.g., new columns, type changes) without rewriting entire tables or complex migrations.
- Setting up a local Spark environment with an Iceberg catalog is straightforward for development and testing, allowing you to prototype lakehouse features efficiently.
- Leveraging a primary key for upserts is essential for ensuring data consistency and idempotency when dealing with frequently updated or late-arriving records.
The Mutable Data Problem
In a perfect world, data would arrive neatly, always conform to its schema, and never need to be updated. But in reality, especially with external APIs, that's rarely the case. Imagine tracking blog posts from an engineering feed: a post might initially appear with a placeholder title, then be updated later. Or a crucial metadata field might be missing in the initial scrape but added in a subsequent fetch. Traditional data lakes, often built on parquet or ORC files, excel at append-only workloads. When you need to update a single record or a small batch, you're usually forced into a painful read-modify-write cycle that involves rewriting entire partitions, leading to inefficient resource usage and complex data consistency challenges. This is where the concept of a "lakehouse" architecture, powered by table formats like Apache Iceberg, truly shines.
Data and Sources
For this demonstration, we're going to ingest data from the GitHub Engineering RSS Feed. This is a dynamic source, with new posts appearing regularly and the potential for existing posts to be updated (though less common for RSS, it's a good proxy for general API mutability). We'll use Python's `feedparser` library to parse the XML feed into a structured format that Spark can consume.
Data accessed on 2024-07-29.
Setting Up Your Local Iceberg Environment
To experiment with Iceberg without a full-blown cluster, I typically start with a local Spark setup configured to use an Iceberg catalog. This allows us to create and manage Iceberg tables directly on the local filesystem. We'll use the `pyspark` library and configure it to include the necessary Iceberg JARs and point to a local Spark warehouse directory.
from pyspark.sql import SparkSession
def get_spark_session(warehouse_path):
"""
Configures and returns a SparkSession with Iceberg support.
"""
spark = SparkSession.builder \
.appName("IcebergApiIngestion") \
.master("local[*]") \
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
.config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkSessionCatalog") \
.config("spark.sql.catalog.spark_catalog.type", "hive") \
.config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog") \
.config("spark.sql.catalog.local.type", "hadoop") \
.config("spark.sql.catalog.local.warehouse", warehouse_path) \
.config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.4.2") \
.getOrCreate()
return spark
Here, I'm setting up two catalogs: `spark_catalog` as a default Hive-compatible catalog (though we won't use Hive itself for this local setup) and `local` as a Hadoop catalog pointing to our local warehouse. The `spark.jars.packages` line is critical; it tells Spark to download the necessary Iceberg runtime JARs automatically. This snippet defines a reusable function, `get_spark_session`, that sets up this environment for us.
Fetching and Structuring API Data
Our `feedparser` step is where we fetch the raw XML and transform it into a more structured format. We need to parse each entry, extract relevant fields like title, link, published date, and a unique identifier. For idempotency, having a stable primary key is paramount. The `link` of a blog post is typically a good candidate for this.
import feedparser
import requests
from datetime import datetime
from pyspark.sql import Row
from pyspark.sql.functions import lit
def fetch_and_parse_feed(url, error_log):
"""
Fetches the RSS feed and parses it into a list of dictionaries.
"""
try:
response = requests.get(url, timeout=10)
response.raise_for_status() # Raise an exception for HTTP errors
feed = feedparser.parse(response.text)
parsed_entries = []
for entry in feed.entries:
# Generate a consistent ID, link is usually stable
entry_id = entry.link
# Ensure published_parsed is handled gracefully
published_dt = datetime(*entry.published_parsed[:6]) if hasattr(entry, 'published_parsed') else
