Skip to content
Home
data-engineering
spark

Taming the Torrent: Idempotent Incremental Ingestion for Dynamic Feeds with PySpark

Mausam

When I first tried to pull the GitHub Engineering blog RSS feed into our lake, the pipeline kept re‑writing the same 10 posts every night and occasionally crashed when a new field appeared in the XML. If you’re a data engineer who needs a repeatable batch job that only writes brand‑new rows, adapts to schema drift, and leaves a clean, auditable Parquet table, this walk‑through is for you. I’ll show you how to turn a volatile RSS feed into an idempotent, schema‑aware PySpark ingestion job that you can schedule on Airflow or Prefect and trust to run forever.

Key Takeaways

  • Use a deterministic primary key (the article link) and a left_anti_join to isolate truly new records on every run.
  • Leverage unionByName and mergeSchema=True to absorb new columns without breaking downstream jobs.
  • Partition by a time‑derived field (e.g., published_date) and write in append mode to keep the lake append‑only while still supporting logical upserts on read.
  • Wrap external HTTP calls and XML parsing in explicit try/except blocks so a single malformed entry never kills the whole batch.
  • Validate idempotency by re‑running the script locally – the row count should stay constant after the first successful load.

The Problem

Dynamic content feeds—RSS, JSON APIs, webhook streams—rarely come with a stable contract. New fields appear, some entries are missing, and the feed may return the same items on every poll. A naïve batch job that blindly appends each fetch ends up with duplicate rows, exploding storage, and downstream analysts spending time de‑duplicating. The goal is a pipeline that:

  1. Fetches the feed reliably.
  2. Transforms it into a flat DataFrame.
  3. Detects only the brand‑new items.
  4. Handles schema changes without manual code edits.
  5. Writes the result to a partitioned Parquet lake in an idempotent way.

Data and Sources

We ingest the public GitHub Engineering RSS feed: https://github.blog/engineering/feed/. Parsing is done with the feedparser library, and all Spark operations rely on the official PySpark API: https://spark.apache.org/docs/latest/api/python/index.html. Freshness note: Data accessed on 2026-10-12.

Loading the Data

The first hurdle is turning XML into a Spark DataFrame while normalizing timestamps and handling missing fields.

import feedparser
from datetime import datetime
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

def fetch_feed(url: str) -> list[dict]:
    try:
        raw = feedparser.parse(url)
    except Exception as e:
        raise RuntimeError(f"Failed to fetch RSS feed: {e}")

    records = []
    for entry in raw.entries:
        # Defensive extraction – missing fields become None
        title = entry.get('title')
        link = entry.get('link')
        summary = entry.get('summary')
        published_raw = entry.get('published')
        # Standardize timestamp
        try:
            published = datetime.strptime(published_raw, "%a, %d %b %Y %H:%M:%S %Z")
        except Exception:
            published = None
        records.append({
            "id": link,               # deterministic primary key
            "title": title,
            "link": link,
            "summary": summary,
            "published": published,
        })
    return records

spark = SparkSession.builder.appName("rss_ingest").getOrCreate()
feed_url = "https://github.blog/engineering/feed/"

raw_records = fetch_feed(feed_url)
current_feed_df = spark.createDataFrame(raw_records)

We now have a flat DataFrame with a reliable id column and a Spark‑inferred schema.

The Core Logic

Step 1 — From XML to DataFrame: Initial Ingestion and Flattening

This step solves the problem of pulling semi‑structured XML and turning it into a tabular form. The code above extracts the five fields we care about, coerces the publish date to a TimestampType, and builds a DataFrame in a single call.

Step 2 — Building for Idempotency: Detecting New Records for Incremental Loads

We need to compare the freshly fetched feed with what we already have in the lake. The primary key is the article URL; using a left_anti_join gives us only the rows that do not exist yet.

import os
from pyspark.sql.functions import to_date, col

OUTPUT_PATH = "s3://my-data-lake/github_engineering/rss_parquet"

def load_existing():
    if os.path.isdir(OUTPUT_PATH):
        return spark.read.parquet(OUTPUT_PATH, mergeSchema=True)
    else:
        # Return an empty DataFrame with the same schema on first run
        return spark.createDataFrame([], current_feed_df.schema)

existing_df = load_existing()

# Identify new records only
new_records_df = current_feed_df.join(existing_df, on="id", how="left_anti")
print(f"New records this run: {new_records_df.count()}")

Because the join is performed on the deterministic id, re‑running the script after a successful load yields zero new rows, proving idempotency.

Step 3 — Evolving Schemas: Adapting to Unforeseen Changes

Feeds can add columns (e.g., author_email) without warning. We simulate a schema change by adding a column to a subset of rows, then merge the two DataFrames with unionByName.

from pyspark.sql.functions import lit

# Simulate schema evolution: add 'author_email' to half of the rows
evolved_df = current_feed_df.withColumn(
    "author_email",
    lit(None).cast(StringType())
).withColumn(
    "author_email",
    col("author_email").when(col("id").rlike("github.com"), "[email protected]")
)

# Union the original and evolved frames – new column is preserved
combined_df = new_records_df.unionByName(evolved_df, allowMissingColumns=True)

When we later read the lake with mergeSchema=True, Spark automatically adds the new column to the existing table.

Step 4 — Efficient Incremental Merges with Data Lake Patterns

We now persist only the truly new rows, partitioned by the date portion of published. This keeps write latency low and enables cheap pruning for downstream queries.

# Add a partition column
final_df = combined_df.withColumn("published_date", to_date(col("published")))

# Append‑only write – idempotency already guaranteed by step 2
final_df.write.mode("append") \
    .partitionBy("published_date") \
    .parquet(OUTPUT_PATH)

# Full logical view (read‑on‑write) – deduplicate just in case
full_df = spark.read.parquet(OUTPUT_PATH, mergeSchema=True) \
    .dropDuplicates(["id"])
print(f"Total rows in lake after merge: {full_df.count()}")

The lake now contains every unique article ever seen, with any newly discovered columns automatically added.

Putting It Together

The snippets above form a coherent pipeline: fetch → normalize → compare → evolve → write. Two error‑handling layers protect us from network hiccups and malformed timestamps.

def main():
    try:
        raw_records = fetch_feed(feed_url)
    except RuntimeError as e:
        print(e)
        return

    current_feed_df = spark.createDataFrame(raw_records)

    existing_df = load_existing()
    new_records_df = current_feed_df.join(existing_df, on="id", how="left_anti")

    if new_records_df.rdd.isEmpty():
        print("No new entries – pipeline is idempotent.")
        return

    # Simulate possible schema evolution (remove in prod)
    evolved_df = current_feed_df.withColumn(
        "author_email", lit(None).cast(StringType())
    )
    combined_df = new_records_df.unionByName(evolved_df, allowMissingColumns=True)

    final_df = combined_df.withColumn("published_date", to_date(col("published")))
    final_df.write.mode("append").partitionBy("published_date").parquet(OUTPUT_PATH)

    full_df = spark.read.parquet(OUTPUT_PATH, mergeSchema=True).dropDuplicates(["id"])
    print(f"Lake now holds {full_df.count()} unique posts.")
    
if __name__ == "__main__":
    main()

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
Idempotent incremental ingestion of the GitHub Engineering RSS feed using PySpark.
"""

import os
import feedparser
from datetime import datetime
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType
from pyspark.sql.functions import lit, col, to_date

FEED_URL = "https://github.blog/engineering/feed/"
OUTPUT_PATH = "s3://my-data-lake/github_engineering/rss_parquet"


def fetch_feed(url: str) -> list[dict]:
    """Fetch RSS XML and return a list of normalized dictionaries."""
    try:
        raw = feedparser.parse(url)
    except Exception as exc:
        raise RuntimeError(f"Failed to fetch RSS feed: {exc}")

    records = []
    for entry in raw.entries:
        title = entry.get("title")
        link = entry.get("link")
        summary = entry.get("summary")
        published_raw = entry.get("published")
        try:
            published = datetime.strptime(published_raw, "%a, %d %b %Y %H:%M:%S %Z")
        except Exception:
            published = None
        records.append(
            {
                "id": link,
                "title": title,
                "link": link,
                "summary": summary,
                "published": published,
            }
        )
    return records


def load_existing(spark):
    """Load existing Parquet lake if present; otherwise return empty DF."""
    if os.path.isdir(OUTPUT_PATH):
        return spark.read.parquet(OUTPUT_PATH, mergeSchema=True)
    else:
        # Empty placeholder – schema will be inferred from first batch
        return spark.createDataFrame([], spark.createDataFrame(fetch_feed(FEED_URL)).schema)


def main():
    spark = SparkSession.builder.appName("rss_ingest").getOrCreate()

    # Step 1: fetch and flatten
    try:
        raw_records = fetch_feed(FEED_URL)
    except RuntimeError as err:
        print(err)
        return

    current_feed_df = spark.createDataFrame(raw_records)

    # Step 2: idempotent detection
    existing_df = load_existing(spark)
    new_records_df = current_feed_df.join(existing_df, on="id", how="left_anti")
    new_count = new_records_df.count()
    print(f"Discovered {new_count} new record(s)")

    if new_count == 0:
        print("Pipeline is idempotent – nothing to write.")
        spark.stop()
        return

    # Step 3: simulate schema evolution (remove in production)
    evolved_df = current_feed_df.withColumn(
        "author_email", lit(None).cast(StringType())
    )
    combined_df = new_records_df.unionByName(evolved_df, allowMissingColumns=True)

    # Step 4: write incrementally, partitioned by date
    final_df = combined_df.withColumn("published_date", to_date(col("published")))
    final_df.write.mode("append").partitionBy("published_date").parquet(OUTPUT_PATH)

    # Verify full logical view
    full_df = spark.read.parquet(OUTPUT_PATH, mergeSchema=True).dropDuplicates(["id"])
    print(f"Lake now contains {full_df.count()} unique posts.")

    spark.stop()


if __name__ == "__main__":
    main()

Expected Output

On the first run you’ll see something like:

Discovered 10 new record(s)
Lake now contains 10 unique posts.

On subsequent runs (assuming no new articles) the script prints:

Discovered 0 new record(s)
Pipeline is idempotent – nothing to write.

Limitations and Tradeoffs

This append‑only + deduplicate‑on‑read pattern works well when the source only adds new rows. If an article is edited after publication, the pipeline will treat it as unchanged because the primary key (URL) matches an existing row; you would need a full merge or a transactional format (Delta Lake/Iceberg) to capture updates. Also, plain Parquet offers no ACID guarantees, so a failure mid‑write could leave a partially written partition. Finally, schema evolution is limited to additive changes – type changes or column renames still require manual migration.

Frequently Asked Questions

How does this compare to using Delta Lake or Iceberg for upserts?

Delta/Iceberg provide native MERGE INTO and atomic commits, eliminating the need for a final dropDuplicates. They also support time‑travel and vacuuming out old versions. The ingestion logic (fetch, key generation, schema handling) stays the same; only the write path changes.

What if the RSS feed doesn’t expose a stable unique identifier?

Compose a composite key from multiple fields (e.g., title + published) or hash the entire entry payload. Be aware that hash collisions are possible but unlikely with a good algorithm (SHA‑256).

How can I monitor schema drift in production?

Persist the inferred schema of each run (e.g., as JSON) and diff it against a baseline. Alert when new columns appear or when a column’s data type changes. Tools like Great Expectations can automate such checks.

Is partitioning by published_date always the best choice?

It works for time‑ordered feeds. If you query primarily by author or tag, consider secondary partitioning or clustering. Over‑partitioning (one day per file) can cause small‑file problems; adjust granularity based on ingestion volume.

How do I handle network failures when fetching the RSS feed?

Wrap the HTTP call in a retry loop (e.g., using tenacity) and surface a clear error code to the orchestrator. In the script above we catch any exception and abort gracefully, preventing a half‑baked load.

What I'd Change

If I were to ship this at scale, I’d replace the raw Parquet lake with Delta Lake. That single change lets me turn the final deduplication step into a true MERGE INTO, gives me atomicity, and opens the door to incremental streaming reads. I’d also externalize the primary‑key logic into a config file so the same pipeline can ingest other RSS sources without code changes. Finally, I’d add a lightweight schema‑registry service to version‑track column additions, making downstream downstream consumers aware of drift before it lands in their notebooks.

Post a Comment

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