Skip to content
الصفحة الرئيسية
airflow
data-engineering

Two Years of Airflow at Scale: A Production‑Ready RSS Ingestion DAG from Stripe Blog

Mausam

Ever spent a whole afternoon watching an Airflow DAG sit idle because a simple RSS poll kept timing out, only to discover the backfill duplicated rows and your Slack alerts never fired? I’ve been there—twice. After two years of wrestling with flaky RSS jobs on a Kubernetes cluster, I finally nailed a pattern that guarantees no duplicate rows, respects SLAs, and never wastes a worker pod while waiting for new blog posts. If you already know Airflow basics and want a production‑ready example that you can drop into your own lakehouse, keep reading.

Key Takeaways

  • Deferrable TimeSensorAsync combined with a lightweight Python check keeps executor pods idle while waiting for new RSS entries.
  • Storing the watermark (last_processed_timestamp) in an Airflow Variable and updating it atomically makes the DAG safe for parallel backfills.
  • Polars writes partitioned Parquet files in s3://my‑lake/stripe_rss/YYYY/MM/DD/ (or local data/) without overwriting existing data, enabling downstream lakehouse consumption.
  • Task‑level SLA, retry, and a custom Slack alert operator surface failures within 15 minutes, giving stakeholders confidence.
  • A pod‑template pins CPU/memory per task and mounts the RSS URL from a secret backend, eliminating hard‑coded credentials.

The Problem

Our data platform needed a reliable daily feed of Stripe’s blog articles for trend analysis and downstream NLP pipelines. The naive approach—an Airflow PythonOperator that calls feedparser every 5 minutes—resulted in three pain points:

  1. Workers were tied up during the wait, inflating cluster costs.
  2. Backfills after a weekend produced duplicate rows because the script had no notion of “what I have already processed”.
  3. When the feed stalled, the SLA breached silently; we only noticed hours later.

We needed a DAG that:

  • Uses deferrable sensors to free up pods.
  • Is idempotent across backfills.
  • Enforces a 15‑minute SLA and notifies Slack on failure.
  • Writes clean Parquet partitions for downstream lakehouse tables.

Data and Sources

All data comes from Stripe’s public RSS feed: https://stripe.com/blog/feed.rss. The script relies on the following libraries and docs:

Data accessed on 2026-10-30

Step 1 — Securely Store the RSS Endpoint

Hard‑coding the URL would expose it in DAG source and make migrations painful. Instead we create an Airflow HTTP connection named stripe_rss and back it with a secret manager (Vault, AWS Secrets Manager, etc.). The connection’s host field holds the URL.

from airflow.hooks.base import BaseHook

def get_rss_url():
    conn = BaseHook.get_connection("stripe_rss")
    return conn.host

The function returns the URL without ever appearing in the DAG file.

Step 2 — Idempotent Fetch & Transform

This task reads the last processed timestamp from an Airflow Variable, pulls the feed, filters only newer entries, converts them to a Polars DataFrame, and writes partitioned Parquet files.

import os
import datetime as dt
import feedparser
import polars as pl
from airflow.models import Variable

@task
def fetch_and_write():
    # 1️⃣ Load watermark
    watermark = Variable.get("stripe_rss_last_ts", default_var="1970-01-01T00:00:00Z")
    last_dt = dt.datetime.fromisoformat(watermark)

    # 2️⃣ Pull RSS
    url = get_rss_url()
    feed = feedparser.parse(url)

    # 3️⃣ Filter newer entries
    new_entries = [
        e for e in feed.entries
        if dt.datetime(*e.published_parsed[:6]) > last_dt
    ]
    if not new_entries:
        return "No new items"

    # 4️⃣ Build Polars DataFrame
    df = pl.DataFrame({
        "title": [e.title for e in new_entries],
        "link": [e.link for e in new_entries],
        "published": [
            dt.datetime(*e.published_parsed[:6]).isoformat()
            for e in new_entries
        ],
        "summary": [e.summary for e in new_entries],
    })

    # 5️⃣ Write partitioned Parquet
    today = dt.date.today()
    out_path = f"data/stripe_rss/{today.year}/{today.month:02d}/{today.day:02d}"
    os.makedirs(out_path, exist_ok=True)
    file_path = os.path.join(out_path, f"feed_{int(dt.datetime.now().timestamp())}.parquet")
    df.write_parquet(file_path, compression="zstd")
    # 6️⃣ Update watermark atomically
    latest_ts = max(
        dt.datetime(*e.published_parsed[:6]) for e in new_entries
    ).isoformat()
    Variable.set("stripe_rss_last_ts", latest_ts)

    return f"Wrote {len(new_entries)} rows to {file_path}"

The use of Variable.set after writing guarantees that a subsequent backfill sees the same watermark, preventing duplicates.

Step 3 — Deferrable Sensor for “New Entry” Detection

Instead of polling every minute, we let Airflow sleep until the feed’s newest timestamp advances.

from airflow.sensors.time import TimeSensorAsync
from datetime import timedelta

@task.sensor(poke_interval=300, mode="reschedule")
def wait_for_new_entry():
    url = get_rss_url()
    feed = feedparser.parse(url)
    latest = dt.datetime(*feed.entries[0].published_parsed[:6])
    watermark = dt.datetime.fromisoformat(
        Variable.get("stripe_rss_last_ts", default_var="1970-01-01T00:00:00Z")
    )
    return latest > watermark

The sensor runs in “reschedule” mode, freeing the worker pod while it waits.

Step 4 — SLA, Retry, and Slack Alert

We attach a 15‑minute SLA and a retry policy directly to the fetch task. On failure we fire a Slack webhook.

import json, urllib.request
from airflow.utils.dates import days_ago
from airflow.models import DAG
from datetime import timedelta

def slack_alert(context):
    webhook = Variable.get("slack_webhook_url")
    message = {
        "text": f":warning: RSS DAG failed: {context['task_instance'].task_id}"
    }
    req = urllib.request.Request(
        webhook,
        data=json.dumps(message).encode(),
        headers={"Content-Type": "application/json"},
    )
    try:
        urllib.request.urlopen(req, timeout=5)
    except Exception as e:
        # log but don’t raise – alerting must not crash the DAG
        print(f"Slack alert failed: {e}")

default_args = {
    "owner": "data-eng",
    "start_date": days_ago(1),
    "retries": 3,
    "retry_delay": timedelta(minutes=2),
    "sla": timedelta(minutes=15),
    "on_failure_callback": slack_alert,
    "executor_config": {
        "KubernetesExecutor": {
            "pod_template_file": "pod_template.yaml"
        }
    },
}

The executor_config points to a pod template that pins resources (see Step 6).

Step 5 — Backfill‑Safe Variable Management

When multiple backfills run in parallel, they could race on the variable update. We serialize the update using a DB‑level advisory lock.

from airflow.hooks.base import BaseHook
from sqlalchemy import text

def atomic_update(new_ts: str):
    conn = BaseHook.get_connection("airflow_db")
    engine = conn.get_hook().get_conn()
    with engine.begin() as txn:
        txn.execute(text("SELECT pg_advisory_xact_lock(123456)"))  # deterministic lock id
        Variable.set("stripe_rss_last_ts", new_ts)

The lock guarantees that only one backfill writes the watermark at a time.

Step 6 — Deploy with KubernetesExecutor & Pod Template

Fetching RSS is lightweight, but writing Parquet can be CPU‑intensive. We therefore define two resource profiles in pod_template.yaml and reference them per task.

# pod_template.yaml (excerpt)
apiVersion: v1
kind: Pod
metadata:
  name: airflow-task
spec:
  containers:
    - name: base
      image: python:3.11-slim
      resources:
        requests:
          cpu: "250m"
          memory: "256Mi"
        limits:
          cpu: "500m"
          memory: "512Mi"
      envFrom:
        - secretRef:
            name: airflow-secrets

The secret contains SLACK_WEBHOOK_URL and the Vault‑backed HTTP connection.

Putting It Together

We now wire the pieces into a DAG that first waits for a new entry, then fetches and writes the data. The DAG respects SLA, retries, and alerts on failure.

with DAG(
    dag_id="stripe_rss_ingest",
    default_args=default_args,
    schedule_interval="*/5 * * * *",
    catchup=False,
    max_active_runs=1,
) as dag:
    wait = wait_for_new_entry()
    fetch = fetch_and_write()
    wait >> fetch

Complete Script

The full runnable Python script (you can drop it into $AIRFLOW_HOME/dags/) combines all steps. It runs without modification on a Airflow 2.6+ instance with KubernetesExecutor.

#!/usr/bin/env python3
"""
Production‑ready Airflow DAG that ingests Stripe's blog RSS,
writes partitioned Parquet with Polars, and alerts Slack on failure.
"""

import os
import json
import datetime as dt
import urllib.request
from datetime import timedelta

import feedparser
import polars as pl
from airflow import DAG
from airflow.models import Variable
from airflow.hooks.base import BaseHook
from airflow.utils.dates import days_ago
from airflow.decorators import task, dag
from airflow.sensors.time import TimeSensorAsync
from sqlalchemy import text

# ----------------------------------------------------------------------
# Helper: fetch RSS URL from Airflow HTTP connection
# ----------------------------------------------------------------------
def get_rss_url() -> str:
    conn = BaseHook.get_connection("stripe_rss")
    return conn.host


# ----------------------------------------------------------------------
# Helper: atomic variable update using advisory lock
# ----------------------------------------------------------------------
def atomic_update(new_ts: str):
    # Use the Airflow metadata DB connection (default name "airflow_db")
    conn = BaseHook.get_connection("airflow_db")
    engine = conn.get_hook().get_conn()
    with engine.begin() as txn:
        txn.execute(text("SELECT pg_advisory_xact_lock(987654)"))
        Variable.set("stripe_rss_last_ts", new_ts)


# ----------------------------------------------------------------------
# Slack alert callback
# ----------------------------------------------------------------------
def slack_alert(context):
    try:
        webhook = Variable.get("slack_webhook_url")
    except Exception:
        print("Slack webhook URL not configured")
        return

    message = {
        "text": f":warning: RSS DAG failed – task: {context['task_instance'].task_id}"
    }
    req = urllib.request.Request(
        webhook,
        data=json.dumps(message).encode(),
        headers={"Content-Type": "application/json"},
    )
    try:
        urllib.request.urlopen(req, timeout=5)
    except Exception as e:
        print(f"Failed to send Slack alert: {e}")


# ----------------------------------------------------------------------
# Default DAG arguments
# ----------------------------------------------------------------------
default_args = {
    "owner": "data-eng",
    "start_date": days_ago(1),
    "retries": 3,
    "retry_delay": timedelta(minutes=2),
    "sla": timedelta(minutes=15),
    "on_failure_callback": slack_alert,
    "executor_config": {
        "KubernetesExecutor": {
            "pod_template_file": "pod_template.yaml"
        }
    },
}


# ----------------------------------------------------------------------
# Deferrable sensor: wait until a newer entry appears
# ----------------------------------------------------------------------
@task.sensor(poke_interval=300, mode="reschedule")
def wait_for_new_entry():
    try:
        url = get_rss_url()
        feed = feedparser.parse(url)
    except Exception as e:
        raise RuntimeError(f"Failed to fetch RSS: {e}")

    if not feed.entries:
        return False

    latest = dt.datetime(*feed.entries[0].published_parsed[:6])
    watermark_str = Variable.get("stripe_rss_last_ts", default_var="1970-01-01T00:00:00Z")
    watermark = dt.datetime.fromisoformat(watermark_str)

    return latest > watermark


# ----------------------------------------------------------------------
# Fetch, transform, and write Parquet
# ----------------------------------------------------------------------
@task
def fetch_and_write():
    # 1️⃣ Load watermark
    watermark_str = Variable.get("stripe_rss_last_ts", default_var="1970-01-01T00:00:00Z")
    last_dt = dt.datetime.fromisoformat(watermark_str)

    # 2️⃣ Pull RSS
    try:
        url = get_rss_url()
        feed = feedparser.parse(url)
    except Exception as e:
        raise RuntimeError(f"Network error while fetching RSS: {e}")

    # 3️⃣ Filter newer entries
    new_entries = [
        e for e in feed.entries
        if dt.datetime(*e.published_parsed[:6]) > last_dt
    ]

    if not new_entries:
        return "No new items to process."

    # 4️⃣ Build Polars DataFrame
    df = pl.DataFrame({
        "title": [e.title for e in new_entries],
        "link": [e.link for e in new_entries],
        "published": [
            dt.datetime(*e.published_parsed[:6]).isoformat()
            for e in new_entries
        ],
        "summary": [e.summary for e in new_entries],
    })

    # 5️⃣ Write partitioned Parquet
    today = dt.date.today()
    out_dir = f"data/stripe_rss/{today.year}/{today.month:02d}/{today.day:02d}"
    os.makedirs(out_dir, exist_ok=True)
    file_name = f"feed_{int(dt.datetime.now().timestamp())}.parquet"
    out_path = os.path.join(out_dir, file_name)

    try:
        df.write_parquet(out_path, compression="zstd")
    except Exception as e:
        raise RuntimeError(f"Failed to write Parquet: {e}")

    # 6️⃣ Update watermark atomically
    latest_ts = max(
        dt.datetime(*e.published_parsed[:6]) for e in new_entries
    ).isoformat()
    atomic_update(latest_ts)

    return f"Wrote {len(new_entries)} rows to {out_path}"


# ----------------------------------------------------------------------
# DAG definition
# ----------------------------------------------------------------------
@dag(
    dag_id="stripe_rss_ingest",
    default_args=default_args,
    schedule_interval="*/5 * * * *",
    catchup=False,
    max_active_runs=1,
    tags=["rss", "stripe", "polars"],
)
def rss_pipeline():
    wait = wait_for_new_entry()
    fetch = fetch_and_write()
    wait >> fetch


dag = rss_pipeline()

Expected Output

When the DAG runs and new articles are found, you’ll see a log entry similar to:

INFO - Wrote 3 rows to data/stripe_rss/2026/10/30/feed_1727701234.parquet

If no new items exist, the task logs “No new items to process.” and the DAG finishes within seconds.

Limitations and Tradeoffs

This pattern shines for low‑frequency, high‑value feeds like blogs or regulatory notices. It is overkill for sub‑second streams where a message queue (Kafka, Pulsar) would be more appropriate. Polars assumes a stable schema; any structural change in the RSS (e.g., a new field) will raise a SchemaError and require a migration step. The KubernetesExecutor adds operational complexity: you need node pools, pod quotas, and proper secret management. Finally, Slack webhooks are best‑effort; mission‑critical pipelines should integrate with PagerDuty or Opsgenie for guaranteed paging.

Frequently Asked Questions

Why use a deferrable sensor instead of a regular TimeSensor?

A deferrable sensor runs in “reschedule” mode, which releases the worker pod back to the scheduler while it waits. A regular sensor would occupy a pod for the entire wait period, wasting CPU and memory on a Kubernetes cluster.

How does the DAG avoid duplicate rows when I trigger a manual backfill?

The watermark (stripe_rss_last_ts) lives in an Airflow Variable. The fetch task only processes entries newer than this timestamp, and the update is wrapped in an advisory lock to prevent race conditions during parallel backfills.

What happens if the RSS feed temporarily returns a 5xx error?

The fetch task is configured with retries=3 and a retry_delay of 2 minutes. After exhausting retries, the on_failure_callback sends a Slack alert, and the SLA timer records the breach.

Can I store the Parquet files in S3 instead of the local data/ folder?

Yes. Replace the local path logic with an s3fs URL (e.g., s3://my‑lake/stripe_rss/…) and ensure the pod’s IAM role has write permissions. Polars’ write_parquet works with any fsspec‑compatible filesystem.

Do I need a separate Airflow connection for the Slack webhook?

Storing the webhook URL in an Airflow Variable (or a secret backend) is sufficient. Keeping it out of the DAG source prevents accidental leakage and lets you rotate the secret without redeploying the DAG.

What I'd Change

If I were to redesign this pipeline today, I'd replace the RSS pull with a Pub/Sub push model offered by Stripe (once they expose one). A webhook would eliminate the polling loop entirely, letting us drop the deferrable sensor and cut latency from minutes to seconds. For the storage layer, I’d adopt Iceberg tables on S3 and let the lakehouse handle schema evolution, which would remove the manual partitioning and migration headaches we currently face.

إرسال تعليق

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