From Cloudflare Blog to Data Lake: Building a Production‑Ready Airflow Pipeline with Polars

When I first tried to surface Cloudflare’s daily blog updates for my analytics team, the naïve “fetch‑transform‑load” script I wrote kept re‑processing the same articles and silently dropped failures during network hiccups. You need a pipeline that guarantees each article is handled exactly once, can survive transient errors, and leaves clean Parquet files for downstream BI or LLM workloads. In this tutorial I show how to wire Airflow 2.7+, Polars 0.20+, and S3 together so the pipeline is safe, observable, and scalable.

Key Takeaways

• Airflow’s PythonOperator + XCom can safely transport Polars DataFrames between tasks.
• Generating a TaskGroup per RSS entry lets you parallelise article‑level processing while keeping the DAG size manageable.
• A tiny Postgres state table enforces idempotency without external lock services.
• Sensors and SlackWebhookOperator give you deterministic start times and instant failure alerts.
• Persisting each article as a Parquet file and then aggregating a daily snapshot creates a reusable data lake asset.

The Problem

Our analytics team needed a reliable daily feed of Cloudflare blog posts. Requirements were:

• Exactly‑once processing – no duplicate rows even if the DAG reruns.
• Automatic retries for transient network or S3 errors.
• Visibility into failures via Slack.
• Scalable handling of a growing number of articles (the feed can contain dozens of items).

Data and Sources

We consume the official Cloudflare RSS endpoint: https://blog.cloudflare.com/rss/. Airflow 2.7+ provides TaskGroup and DAG serialization. Polars 0.20+ offers fast columnar transforms. PostgreSQL stores the idempotency table (you can swap SQLite for a dev environment). S3 (or MinIO) is the final storage layer.
Data accessed on 2026-10-29.

Step 1 — Set Up Airflow Connections & Variables

Airflow stores external credentials as Connections and runtime constants as Variables. This step isolates secrets from code.

# In the Airflow UI or via CLI
# Variable holding the RSS URL
airflow variables set cloudflare_rss_url "https://blog.cloudflare.com/rss/"

# Connection for S3 (aws_default)
#   - Type: Amazon Web Services
#   - Login: YOUR_AWS_ACCESS_KEY
#   - Password: YOUR_AWS_SECRET_KEY
#   - Extra: {"region_name": "us-east-1"}

# Connection for Postgres (postgres_default)
#   - Host, Schema, Login, Password as usual

Step 2 — PythonOperator to Fetch & Parse RSS

The first task pulls the XML feed, parses it with feedparser, and pushes the raw list of entries to XCom.

def fetch_rss(**context):
    import feedparser
    from airflow.models import Variable
    url = Variable.get("cloudflare_rss_url")
    try:
        feed = feedparser.parse(url)
        if feed.bozo:
            raise ValueError("Malformed RSS")
        entries = feed.entries
        # Push as XCom (JSON‑serialisable)
        context["ti"].xcom_push(key="raw_entries", value=entries)
    except Exception as e:
        raise RuntimeError(f"Failed to fetch RSS: {e}")

fetch_task = PythonOperator(
    task_id="fetch_rss",
    python_callable=fetch_rss,
    retries=3,
    retry_delay=timedelta(minutes=5),
)

Step 3 — Polars Transformation & Deduplication

We convert the list of dicts into a Polars DataFrame, keep only the columns we need, and drop duplicates by link. The DF is serialized to bytes before being stored in XCom (Polars implements to_arrow() which XCom can handle).

def transform_entries(**context):
    import polars as pl
    raw = context["ti"].xcom_pull(key="raw_entries", task_ids="fetch_rss")
    df = pl.DataFrame(raw).select([
        pl.col("title"),
        pl.col("link"),
        pl.col("published"),
        pl.col("summary"),
    ])
    df = df.unique(subset=["link"])
    # Serialize to Arrow bytes for XCom transport
    bytes_blob = df.to_arrow().to_pybytes()
    context["ti"].xcom_push(key="clean_df", value=bytes_blob)

transform_task = PythonOperator(
    task_id="transform",
    python_callable=transform_entries,
    retries=2,
    retry_delay=timedelta(minutes=2),
)

Step 4 — Dynamic Task Generation for Each Entry

After transformation we spin up a TaskGroup. Each article gets its own PythonOperator that writes a Parquet file to S3. This parallelism scales with the number of entries while keeping the DAG definition concise.

def create_article_tasks(parent_dag, df_bytes):
    import polars as pl
    from airflow.operators.python import PythonOperator
    from airflow.utils.task_group import TaskGroup
    from airflow.providers.amazon.aws.hooks.s3 import S3Hook

    df = pl.from_arrow(pl.read_ipc(df_bytes))
    tg = TaskGroup(group_id="process_articles")
    for row in df.iter_rows():
        title, link, published, summary = row

        def write_parquet(**context, link=link, title=title,
                          published=published, summary=summary):
            import io, pyarrow.parquet as pq
            import polars as pl
            s3 = S3Hook(aws_conn_id="aws_default")
            # Idempotency check (see Step 5)
            if check_processed(link):
                return
            article_df = pl.DataFrame({
                "title": [title],
                "link": [link],
                "published": [published],
                "summary": [summary],
            })
            buffer = io.BytesIO()
            article_df.write_parquet(buffer)
            buffer.seek(0)
            key = f"cloudflare/{published[:10]}/{link.split('/')[-1]}.parquet"
            s3.load_bytes(buffer.read(), bucket_name="my-bucket", key=key)
            mark_processed(link)

        PythonOperator(
            task_id=f"write_{hash(link)}",
            python_callable=write_parquet,
            retries=3,
            retry_delay=timedelta(minutes=3),
        ).set_upstream(parent_dag.get_task("transform"))
    return tg

article_group = PythonOperator(
    task_id="generate_article_tasks",
    python_callable=lambda **ctx: create_article_tasks(ctx["dag"], 
        ctx["ti"].xcom_pull(key="clean_df", task_ids="transform")),
)

Step 5 — Idempotency State Table

A lightweight Postgres table records every processed link. Before writing we query the table; after a successful write we insert the link. This guarantees exactly‑once semantics even if the DAG is manually cleared.

def get_pg_hook():
    from airflow.providers.postgres.hooks.postgres import PostgresHook
    return PostgresHook(postgres_conn_id="postgres_default")

def check_processed(link: str) -> bool:
    hook = get_pg_hook()
    sql = "SELECT 1 FROM processed_entries WHERE link = %s"
    result = hook.get_first(sql, parameters=(link,))
    return result is not None

def mark_processed(link: str):
    hook = get_pg_hook()
    sql = "INSERT INTO processed_entries (link) VALUES (%s) ON CONFLICT DO NOTHING"
    hook.run(sql, parameters=(link,))

Step 6 — Retry Logic & Sensors

We want the pipeline to start after Cloudflare’s daily update window (around 5 AM UTC). A TimeSensor blocks execution until that hour. All critical tasks already have retries and retry_delay set.

from airflow.sensors.time_sensor import TimeSensor
wait_until_five = TimeSensor(
    task_id="wait_until_five",
    target_time=datetime.time(hour=5, minute=0),
    poke_interval=300,
)

# DAG wiring
wait_until_five >> fetch_task >> transform_task >> article_group

Step 7 — Monitoring & Alerting

Any task failure triggers a Slack webhook. Additionally we log the number of new articles processed at the end of the day.

from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator

alert = SlackWebhookOperator(
    task_id="slack_alert",
    http_conn_id="slack_default",
    webhook_token="YOUR_SLACK_WEBHOOK",
    message="⚠️ Cloudflare RSS pipeline failed!",
    trigger_rule="one_failed",
)

transform_task >> alert

Step 8 — Final Aggregation & Downstream Trigger

When all article tasks finish, we concatenate the per‑article Parquet files into a single daily snapshot. Downstream BI tools can read this single file, and LLM pipelines can ingest it in bulk.

def aggregate_snapshot(**context):
    import polars as pl
    from airflow.providers.amazon.aws.hooks.s3 import S3Hook
    s3 = S3Hook(aws_conn_id="aws_default")
    prefix = f"cloudflare/{datetime.date.today().isoformat()}/"
    keys = s3.list_keys(bucket_name="my-bucket", prefix=prefix)
    dfs = []
    for key in keys:
        obj = s3.get_key(key, bucket_name="my-bucket")
        df = pl.read_parquet(obj)
        dfs.append(df)
    if dfs:
        daily_df = pl.concat(dfs)
        out_key = f"cloudflare/daily_{datetime.date.today().strftime('%Y%m%d')}.parquet"
        buffer = io.BytesIO()
        daily_df.write_parquet(buffer)
        buffer.seek(0)
        s3.load_bytes(buffer.read(), bucket_name="my-bucket", key=out_key)

aggregate_task = PythonOperator(
    task_id="aggregate_snapshot",
    python_callable=aggregate_snapshot,
    retries=2,
    retry_delay=timedelta(minutes=2),
)

article_group >> aggregate_task

Limitations and Tradeoffs

Cloudflare’s RSS endpoint has no published rate limit; aggressive polling could get throttled, so a back‑off strategy is advisable. Loading the entire feed into a Polars DataFrame works for the current ~50‑item daily size, but a feed that grows to thousands of items would require streaming or chunked processing to keep RAM usage low. The Postgres idempotency table grows linearly with processed links; for multi‑year archives you should add a TTL column and periodically purge old rows. Dynamic task generation creates a separate Airflow task per article; if the feed ever exceeds a few hundred items you may need to batch articles into groups to avoid overwhelming the scheduler.

Frequently Asked Questions

Can I replace PostgreSQL with SQLite?

Yes. For a dev or low‑volume environment SQLite works fine; just change the connection ID and ensure the DAG runs on a single worker so the file isn’t contested.

What if an article’s content is larger than the Polars in‑memory limit?

Polars can handle gigabyte‑scale data, but the RSS summary field is short. If you later pull full article bodies, switch to a streaming write (e.g., write each article directly to S3 without loading all rows into a DataFrame).

How do I test the DAG locally without a full Airflow deployment?

Use airflow tasks test to run each operator in isolation, and set AIRFLOW__CORE__UNIT_TEST_MODE=True to bypass the scheduler.

Is the Slack alert enough for production monitoring?

Slack gives immediate human notification, but you should also enable Airflow’s built‑in metrics (Prometheus or StatsD) and set up alerting on task failure rates.

Can I run this pipeline on a KubernetesExecutor?

All operators are container‑agnostic; just ensure the Docker image includes Python, Polars, feedparser, and the AWS/Postgres drivers.

What I'd Change

If I were to redesign this pipeline for a multi‑tenant data platform, I’d replace the per‑article TaskGroup with a single “batch processor” that reads the raw RSS entries, deduplicates, and writes a batched Parquet file in one go. That eliminates the scheduler overhead of dozens of tiny tasks and lets you leverage Polars’ parallel I/O directly. For the current use‑case—fine‑grained visibility and per‑article error isolation—the dynamic tasks win, but scaling beyond a few hundred entries calls for the batch approach.

Complete Script

The full runnable script combining all steps. Save as cloudflare_rss_dag.py and drop it into your Airflow dags/ folder.

#!/usr/bin/env python3
"""
Airflow DAG: Cloudflare RSS → Polars → S3 with exactly‑once semantics.
"""

import datetime
import io
import json
import time
from datetime import timedelta

import feedparser
import polars as pl
from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from airflow.sensors.time_sensor import TimeSensor
from airflow.utils.task_group import TaskGroup

# ------------------- Helper Functions ------------------- #
def get_pg_hook():
    return PostgresHook(postgres_conn_id="postgres_default")

def check_processed(link: str) -> bool:
    hook = get_pg_hook()
    sql = "SELECT 1 FROM processed_entries WHERE link = %s"
    return hook.get_first(sql, parameters=(link,)) is not None

def mark_processed(link: str):
    hook = get_pg_hook()
    sql = """
    INSERT INTO processed_entries (link) VALUES (%s)
    ON CONFLICT (link) DO NOTHING
    """
    hook.run(sql, parameters=(link,))

# ------------------- DAG Definition ------------------- #
default_args = {
    "owner": "mausam",
    "depends_on_past": False,
    "email_on_failure": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="cloudflare_rss_to_lake",
    start_date=datetime.datetime(2026, 10, 1),
    schedule_interval="@daily",
    catchup=False,
    default_args=default_args,
    max_active_tasks=20,
) as dag:

    # 1. Wait until 5 AM UTC (Cloudflare update window)
    wait_until_five = TimeSensor(
        task_id="wait_until_five",
        target_time=datetime.time(hour=5, minute=0),
        poke_interval=300,
    )

    # 2. Fetch RSS
    def fetch_rss(**context):
        url = Variable.get("cloudflare_rss_url")
        try:
            feed = feedparser.parse(url)
            if feed.bozo:
                raise ValueError("Malformed RSS")
            entries = feed.entries
            context["ti"].xcom_push(key="raw_entries", value=entries)
        except Exception as exc:
            raise RuntimeError(f"RSS fetch failed: {exc}")

    fetch_task = PythonOperator(
        task_id="fetch_rss",
        python_callable=fetch_rss,
        retries=3,
        retry_delay=timedelta(minutes=5),
    )

    # 3. Transform with Polars
    def transform_entries(**context):
        raw = context["ti"].xcom_pull(key="raw_entries", task_ids="fetch_rss")
        df = pl.DataFrame(raw).select([
            pl.col("title"),
            pl.col("link"),
            pl.col("published"),
            pl.col("summary"),
        ])
        df = df.unique(subset=["link"])
        # Serialize Arrow bytes for XCom
        bytes_blob = df.to_arrow().to_pybytes()
        context["ti"].xcom_push(key="clean_df", value=bytes_blob)

    transform_task = PythonOperator(
        task_id="transform",
        python_callable=transform_entries,
        retries=2,
        retry_delay=timedelta(minutes=2),
    )

    # 4. Dynamic article tasks
    def generate_article_tasks(**context):
        df_bytes = context["ti"].xcom_pull(key="clean_df", task_ids="transform")
        df = pl.from_arrow(pl.read_ipc(df_bytes))
        tg = TaskGroup(group_id="process_articles")
        for row in df.iter_rows():
            title, link, published, summary = row

            def write_parquet(**inner_ctx, link=link, title=title,
                              published=published, summary=summary):
                # Idempotency check
                if check_processed(link):
                    return
                article_df = pl.DataFrame({
                    "title": [title],
                    "link": [link],
                    "published": [published],
                    "summary": [summary],
                })
                buffer = io.BytesIO()
                article_df.write_parquet(buffer)
                buffer.seek(0)
                s3 = S3Hook(aws_conn_id="aws_default")
                key = f"cloudflare/{published[:10]}/{link.split('/')[-1]}.parquet"
                try:
                    s3.load_bytes(buffer.read(), bucket_name="my-bucket", key=key)
                except Exception as exc:
                    raise RuntimeError(f"S3 write failed: {exc}")
                mark_processed(link)

            PythonOperator(
                task_id=f"write_{hash(link)}",
                python_callable=write_parquet,
                retries=3,
                retry_delay=timedelta(minutes=3),
            ).set_upstream(transform_task)
        return tg

    generate_task = PythonOperator(
        task_id="generate_article_tasks",
        python_callable=generate_article_tasks,
    )

    # 5. Aggregate daily snapshot
    def aggregate_snapshot(**context):
        s3 = S3Hook(aws_conn_id="aws_default")
        today = datetime.date.today().isoformat()
        prefix = f"cloudflare/{today}/"
        keys = s3.list_keys(bucket_name="my-bucket", prefix=prefix) or []
        dfs = []
        for key in keys:
            obj = s3.get_key(key, bucket_name="my-bucket")
            df = pl.read_parquet(obj)
            dfs.append(df)
        if not dfs:
            return
        daily_df = pl.concat(dfs)
        out_key = f"cloudflare/daily_{datetime.date.today().strftime('%Y%m%d')}.parquet"
        buffer = io.BytesIO()
        daily_df.write_parquet(buffer)
        buffer.seek(0)
        s3.load_bytes(buffer.read(), bucket_name="my-bucket", key=out_key)

    aggregate_task = PythonOperator(
        task_id="aggregate_snapshot",
        python_callable=aggregate_snapshot,
        retries=2,
        retry_delay=timedelta(minutes=2),
    )

    # 6. Slack alert on any failure
    slack_alert = SlackWebhookOperator(
        task_id="slack_alert",
        http_conn_id="slack_default",
        webhook_token="YOUR_SLACK_WEBHOOK",
        message="⚠️ Cloudflare RSS pipeline failed!",
        trigger_rule="one_failed",
    )

    # DAG wiring
    wait_until_five >> fetch_task >> transform_task >> generate_task >> aggregate_task
    [fetch_task, transform_task, generate_task, aggregate_task] >> slack_alert

Expected Output

When the DAG runs successfully you’ll see:

• Airflow logs confirming each article was written to s3://my-bucket/cloudflare/YYYY-MM-DD/….
• A daily Parquet snapshot at s3://my-bucket/cloudflare/daily_YYYYMMDD.parquet.
• In Slack, a quiet run (no message) or a failure alert with the error traceback.