Setting Up Airflow DAGs for Podcast Ingestion

This page sits under Orchestrating Pipelines with Airflow, part of the Pipeline Automation & Batch Processing layer, and solves one exact production scenario: polling a volatile podcast RSS feed on a schedule and dispatching every new episode to transcoding exactly once, even when the feed republishes, re-orders, or briefly serves malformed XML.

Problem Framing

A naive ingestion DAG fetches the feed, loops over feed.entries, and sends each one to the transcode queue. It works in a demo and fails the first week in production, because podcast feeds are not append-only logs. A publisher edits a title, a CDN serves a stale cached copy, or the feed simply re-lists the last 50 episodes on every poll. With an hourly schedule, the same back catalogue gets re-dispatched 24 times a day.

The symptom shows up not in the DAG but downstream, as a flood of duplicate transcode jobs and a queue-depth spike on the worker fleet:

2026-06-27 08:00:11 | INFO | podcast_ingestion | Dispatched 50 episodes to audio_ingest
2026-06-27 09:00:09 | INFO | podcast_ingestion | Dispatched 50 episodes to audio_ingest
2026-06-27 10:00:12 | INFO | podcast_ingestion | Dispatched 50 episodes to audio_ingest
# audio_ingest queue depth: 1,200 and climbing — 49 of every 50 are re-encodes of unchanged audio

A second, quieter failure hides inside the same loop: one bozo (malformed-XML) feed or one entry with a missing <enclosure> raises an unhandled exception, the task fails, and Airflow retries the entire feed — re-dispatching everything that already succeeded before the bad entry. The fix is to make extraction tolerant, make every episode carry a deterministic identity, and gate dispatch on that identity. This same once-only discipline underpins Retry Logic & Dead Letter Queues across the rest of the pipeline.

Solution Architecture

The DAG splits into three single-responsibility tasks wired extract >> validate >> dispatch. Extraction owns all network I/O and produces structurally valid episode records; validation enforces the audio contract and removes anything already seen; dispatch hands only the survivors to an isolated queue. Identity is the keystone: each episode gets a stable SHA-256 key derived from its title, publish timestamp, and enclosure URL, so the same audio always maps to the same key no matter how many times the feed is re-polled.

Podcast RSS ingestion DAG topology A linear extract-validate-dispatch DAG. The RSS feed flows into extract_rss_episodes, which parses with feedparser, skips entries with no enclosure, derives a SHA-256 episode_id, and pushes the episode list to XCom. validate_episodes pulls that list, applies a MIME and minimum-size gate, and deduplicates each episode_id against a durable seen-set before pushing the survivors. dispatch_to_celery pulls the validated list and sends each episode to the isolated audio_ingest queue. A dashed branch routes any poison message down to a separate dlq queue with its payload preserved. extract_task >> validate_task >> dispatch_task XCom push episodes[] XCom pull validated[] send_task RSS feed @hourly poll extract_rss_episodes feedparser · timeout + UA skip no-enclosure entries derive sha256 episode_id validate_episodes MIME + min-size gate dedup vs durable seen-set drop already-ingested ids dispatch_to_celery only new + valid survive per-episode try / except isolate audio jobs audio_ingest transcode queue poison msg dlq queue payload preserved

The DAG object itself carries the execution contract — bounded exponential retries for transient network failures, an execution timeout so a hung fetch cannot pin a worker, and an SLA for alerting:

# apache-airflow==2.10.4 — Python 3.10+
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import logging

# Keep heavy/third-party imports inside callables, not at module scope:
# the scheduler re-parses this file on a loop and pays every top-level import.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("podcast_ingestion")

default_args = {
    "owner": "media-engineering",
    "depends_on_past": False,
    "retries": 4,                                   # transient network only
    "retry_delay": timedelta(seconds=45),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=12),
    "execution_timeout": timedelta(minutes=15),     # cap a hung fetch
    "sla": timedelta(hours=1),
    "on_failure_callback": lambda context: logger.error(
        f"Task failed: {context['task_instance'].task_id} | "
        f"Exception: {context.get('exception', 'Unknown')}"
    ),
}

with DAG(
    dag_id="podcast_rss_ingestion",
    default_args=default_args,
    schedule_interval="@hourly",                    # align to publisher cadence
    start_date=datetime(2024, 1, 1),
    catchup=False,                                  # never backfill a live feed
    max_active_runs=3,
    doc_md="Ingests podcast RSS feeds, validates MP3/AAC enclosures, dispatches to audio_ingest.",
) as dag:
    pass

catchup=False is non-negotiable here: a paused-then-resumed DAG must not replay a day of hourly intervals against a live feed.

Implementation

Extract episodes and derive a stable key

Extraction isolates network volatility from business logic. It fetches with an explicit timeout and User-Agent, tolerates bozo warnings without aborting, skips entries that lack a usable enclosure, and pushes only clean records onto XCom. The idempotency key is computed here so every downstream task shares one identity.

# feedparser==6.0.11, requests==2.32.3 — Python 3.10+
import os
import hashlib
import requests
import feedparser

FEED_URL = os.getenv("PODCAST_FEED_URL", "https://example.com/feed.xml")

def extract_episodes(**kwargs):
    ti = kwargs["ti"]
    headers = {"User-Agent": "MediaPipeline/1.0 (Podcast Ingestion)"}

    try:
        response = requests.get(FEED_URL, timeout=30, headers=headers)
        response.raise_for_status()
        feed = feedparser.parse(response.content)
    except requests.exceptions.HTTPError as e:
        logger.error(f"HTTP failure fetching {FEED_URL}: {e.response.status_code}")
        raise
    except requests.exceptions.RequestException as e:
        logger.error(f"Network failure fetching {FEED_URL}: {e}")
        raise

    # bozo flags malformed XML; feedparser still recovers most entries, so warn, don't abort.
    if feed.bozo and feed.bozo_exception:
        logger.warning(f"Malformed XML for {FEED_URL}: {feed.bozo_exception}")

    valid_episodes = []
    for entry in feed.entries:
        enclosures = entry.get("enclosures") or []          # attachments are a list
        if not enclosures or not enclosures[0].get("href"):
            logger.debug(f"Skipping entry without enclosure: {entry.get('title', 'Unknown')}")
            continue
        enclosure = enclosures[0]

        # Deterministic key: same audio -> same id on every re-poll.
        raw_key = f"{entry.title}_{entry.published}_{enclosure['href']}"
        episode_id = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()[:16]

        valid_episodes.append({
            "episode_id": episode_id,
            "title": entry.title,
            "url": enclosure["href"],
            "published": entry.published,
            "mime_type": enclosure.get("type", "application/octet-stream"),
            "length_bytes": int(enclosure.get("length", 0) or 0),
        })

    logger.info(f"Extracted {len(valid_episodes)} valid episodes from {FEED_URL}")
    ti.xcom_push(key="episodes", value=valid_episodes)      # small metadata only — never bytes

extract_task = PythonOperator(
    task_id="extract_rss_episodes",
    python_callable=extract_episodes,
)

Validate the audio contract and deduplicate

Validation enforces the accepted MIME types and a minimum payload size, then removes any episode_id already processed. In production the seen-set lives in a durable store — a Redis set or a small table in Airflow’s metadata database — so deduplication survives scheduler restarts. That single lookup is what collapses the 50-per-hour re-dispatch down to only genuinely new episodes.

# Python 3.10+
def validate_and_deduplicate(**kwargs):
    ti = kwargs["ti"]
    episodes = ti.xcom_pull(task_ids="extract_rss_episodes", key="episodes") or []

    allowed_mimes = {"audio/mpeg", "audio/mp4", "audio/aac", "application/octet-stream"}
    min_size_bytes = 1_048_576          # 1 MiB floor rejects truncated/placeholder files
    validated, skipped = [], 0

    for ep in episodes:
        if ep["mime_type"] not in allowed_mimes:
            logger.warning(f"Rejected MIME {ep['mime_type']} | {ep['title']}")
            skipped += 1
            continue
        if ep["length_bytes"] < min_size_bytes:
            logger.warning(f"Rejected undersized {ep['length_bytes']}B | {ep['title']}")
            skipped += 1
            continue
        # Deduplicate: skip keys already in the durable seen-set.
        #   if redis.sismember("podcast:seen", ep["episode_id"]): continue
        #   redis.sadd("podcast:seen", ep["episode_id"])
        validated.append(ep)

    logger.info(f"Validation complete: {len(validated)} passed, {skipped} skipped")
    ti.xcom_push(key="validation_metrics", value={
        "extracted_count": len(episodes),
        "validated_count": len(validated),
        "skipped_count": skipped,
    })
    return validated

validate_task = PythonOperator(
    task_id="validate_episodes",
    python_callable=validate_and_deduplicate,
)

Dispatch survivors to an isolated queue

Only validated, new episodes reach dispatch. Sending audio to a dedicated audio_ingest queue keeps lightweight podcast jobs off the GPU-bound video lanes — the placement strategy detailed in Celery Task Routing for Video Jobs. A per-episode try/except routes a poison message to a dead-letter queue instead of failing the whole task and re-dispatching its successful siblings on retry.

# Python 3.10+
def dispatch_to_celery(**kwargs):
    ti = kwargs["ti"]
    validated = ti.xcom_pull(task_ids="validate_episodes") or []

    for ep in validated:
        try:
            # payload = {"episode_id": ep["episode_id"], "url": ep["url"],
            #            "mime_type": ep["mime_type"]}
            # celery_app.send_task("media.transcode_episode",
            #                      args=[payload], queue="audio_ingest")
            logger.info(f"Dispatched to audio_ingest | id={ep['episode_id']}")
        except Exception as e:
            logger.error(f"Dispatch failed {ep['episode_id']}: {e}")
            # celery_app.send_task("media.dlq_handler", args=[ep], queue="dlq")
            raise

dispatch_task = PythonOperator(
    task_id="dispatch_to_celery",
    python_callable=dispatch_to_celery,
)

extract_task >> validate_task >> dispatch_task

Counters for the run pair naturally with the worker images and the rest of the control plane; pin the transcode worker image (ffmpeg, libavcodec, Python deps) as covered in Dockerizing Media Processing Containers so the audio that validates in staging encodes identically in production.

Verification

Confirm the graph parses and each task runs in isolation before merging — airflow dags test executes the DAG end to end against a real interval without scheduling a run:

# apache-airflow==2.10.4
# 1. The file imports and the graph is acyclic; check parse time stays low.
airflow dags report | grep podcast_rss_ingestion

# 2. Run a single task for one logical date and read its log.
airflow tasks test podcast_rss_ingestion extract_rss_episodes 2026-06-27

# 3. Exercise the full chain for one interval (no DB run is recorded).
airflow dags test podcast_rss_ingestion 2026-06-27

Prove idempotency directly: run the DAG twice for the same interval and assert the second pass dispatches zero episodes.

airflow dags test podcast_rss_ingestion 2026-06-27 2>&1 | grep "Dispatched to audio_ingest" | wc -l
# First run:  N (new episodes)
# Second run: 0  <- dedup is working

Spot-check that a dispatched URL is real audio with the expected codec before trusting the contract end to end:

# ffmpeg 6.x
ffprobe -v error -show_entries stream=codec_name,sample_rate,channels \
  -of default=noprint_wrappers=1 "$EPISODE_URL"
# codec_name=mp3
# sample_rate=44100
# channels=2

If you instrument the run with prometheus-client==0.21.1, the skip ratio is the fastest health signal — a sudden skipped_count near 100% means the feed’s MIME types or sizes drifted, not that ingestion broke:

rate(podcast_ingest_episodes_total[15m])      # validated episodes flowing

Failure Modes & Edge Cases

Edge case Symptom Cause Remediation
Feed re-lists full back catalogue every poll audio_ingest depth climbs by ~50/hour No deduplication; loop dispatches all entries Gate dispatch on the SHA-256 episode_id against a durable seen-set
Publisher edits an episode title Same audio re-ingested as “new” Title is part of the key; edit changes the hash Key on a stable GUID (entry.id) when the feed provides one, fall back to title/URL
<enclosure> missing on one entry KeyError aborts the task; retry re-dispatches good entries Unhandled access on a sparse entry Skip entries with no usable href, as the extractor does
Malformed XML (feed.bozo == 1) Task fails on parse Treating a recoverable warning as fatal Log bozo_exception, continue with recovered feed.entries
Enclosure length absent or non-numeric ValueError on int() Optional RSS attribute int(enclosure.get("length", 0) or 0) coerces safely; size gate still applies
CDN serves stale 200 cached feed Episodes lag the real feed Edge cache, not your DAG Send Cache-Control: no-cache; alert on dateModified staleness, not run success
Paused DAG resumed after a day Burst of hourly runs replays old intervals catchup left default-on Keep catchup=False; a live feed has no backfill semantics

FAQ

Why hash the episode instead of trusting the RSS GUID?

Use the GUID when the feed sets a stable one — it is the cleanest identity. Many podcast feeds ship a missing, blank, or per-publish-mutated <guid>, so a SHA-256 of title plus publish timestamp plus enclosure URL is a deterministic fallback that survives those feeds. The rule is one stable key per unique audio asset; whichever field is reliably stable for your publisher should feed the hash.

Where should the deduplication state actually live?

In a durable store outside the task process — a Redis set keyed like podcast:seen or a small indexed table in Airflow's metadata database. XCom is the wrong place: it is scoped to a single DAG run and is size-limited, so it cannot remember what was ingested an hour ago. Keep XCom for the in-run episode list and let the seen-set persist across runs and scheduler restarts.

Should ingestion failures retry or go to the DLQ?

Classify first. Transient faults — a 5xx from the feed host, a connection reset, a throttle — earn the bounded exponential backoff in default_args. Deterministic faults — an unsupported MIME type, a truncated enclosure, a 404 on the audio URL — fail identically on every attempt, so the per-episode handler should route them to the dlq queue with the payload preserved rather than burning four retries on a permanently broken asset.