How I Added Adaptive Data Quality Checks to a DVC‑Managed RSS Pipeline

Ever pulled an RSS feed only to discover that a new “author” field appeared overnight, breaking every downstream notebook you wrote last week? I hit that exact wall while building a nightly ingest of the Stripe Blog RSS for a sentiment‑analysis prototype. I already versioned the CSV with DVC (see my earlier post on reproducible pipelines), but the data kept slipping through unnoticed until the model crashed. In this tutorial I’ll show you how I turned that fragile pipeline into a self‑healing, version‑aware quality gate using Great Expectations, all while keeping DVC as the source‑of‑truth.

Key Takeaways

• Great Expectations suites can be generated dynamically from a “profile” of the latest data, letting you adapt to schema drift without manual rewrites.
• Versioning both the raw CSV and the expectation suite with DVC gives you a reproducible audit trail of data quality over time.
• Embedding the validation step in a CI‑friendly checkpoint makes a failing quality check stop the pipeline before bad data propagates.

The Problem

My earlier DVC pipeline reliably stored stripe_blog_posts.csv in the data lake, but it had no guardrails. A single missing published timestamp or an extra category column caused downstream parsing errors. I needed a way to (1) automatically discover the current shape of the feed, (2) codify business rules (e.g., every record must have a URL and a non‑empty title), and (3) keep those rules versioned alongside the data so I could trace regressions.

Data and Sources

The live feed lives at https://stripe.com/blog/feed.rss. I use feedparser to turn XML into Python dicts, pandas for tabular conversion, Great Expectations for validation, and DVC for artifact versioning. Data accessed on 2026-11-03.

Step 1 — Ingesting the RSS Feed

The first hurdle is turning the semi‑structured XML into a clean CSV that DVC can track. I wrap the fetch in a try/except to surface network problems early.

import feedparser
import pandas as pd
import requests

def fetch_rss(url: str) -> pd.DataFrame:
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
    except requests.RequestException as e:
        raise RuntimeError(f"Failed to download RSS: {e}") from e

    feed = feedparser.parse(resp.text)
    entries = feed.entries
    # Normalise to a flat table
    rows = []
    for e in entries:
        rows.append({
            "title": e.get("title", "").strip(),
            "link": e.get("link", ""),
            "published": e.get("published", ""),
            "author": e.get("author", ""),
            "tags": ", ".join([t.term for t in e.get("tags", [])]),
        })
    return pd.DataFrame(rows)

With the DataFrame ready, I dump it to data/stripe_blog_posts.csv and let DVC track the file.

Step 2 — Building a Baseline Expectation Suite

Great Expectations can “profile” a batch and emit a JSON suite that reflects the current schema. I run this once after the first successful ingest to seed the suite.

import great_expectations as ge
from great_expectations.checkpoint import Checkpoint
from great_expectations.core.batch import BatchRequest

def create_expectation_suite(df: pd.DataFrame, suite_name: str) -> None:
    context = ge.get_context()
    # Register the CSV as a datasource (once per project)
    if "stripe_rss" not in context.list_datasources():
        context.add_datasource(
            "stripe_rss",
            class_name="PandasDatasource",
        )
    # Save the DataFrame as a temporary batch
    batch = context.get_batch({
        "datasource_name": "stripe_rss",
        "data_asset_name": "stripe_blog_posts",
        "batch_data": df,
    })
    suite = context.create_expectation_suite(suite_name, overwrite=True)
    batch.expect_table_row_count_to_be_between(min_value=1, max_value=1000)
    batch.expect_column_to_exist("title")
    batch.expect_column_values_to_not_be_null("title")
    batch.expect_column_to_exist("link")
    batch.expect_column_values_to_match_regex("link", r"^https?://")
    batch.expect_column_to_exist("published")
    # Dynamically add a type check based on the first row
    sample_date = df["published"].iloc[0]
    if sample_date:
        batch.expect_column_values_to_match_strftime_format("published", "%a, %d %b %Y %H:%M:%S %z")
    batch.save_expectation_suite(discard_failed_expectations=False)
    context.save_expectation_suite(suite, suite_name=suite_name)

This suite lives under great_expectations/expectations/ and is automatically versioned by DVC because the entire great_expectations folder is tracked.

Step 3 — Adding Business‑Specific Rules

Beyond generic schema checks I need a rule that “tags” must contain at least one comma‑separated term. I encode it as a custom expectation.

def add_custom_expectations(context, suite_name):
    suite = context.get_expectation_suite(suite_name)
    suite.add_expectation({
        "expectation_type": "expect_column_values_to_match_regex",
        "kwargs": {
            "column": "tags",
            "regex": r".+",
            "mostly": 0.9,
            "result_format": "BASIC",
        },
        "meta": {"notes": "At least one tag per post"},
    })
    context.save_expectation_suite(suite, suite_name=suite_name)

Step 4 — Integrating Validation with DVC

The final piece is a DVC stage that runs the checkpoint after the CSV is produced. The checkpoint aborts with a non‑zero exit code if any expectation fails, causing the DVC pipeline to stop.

def run_validation(csv_path: str, suite_name: str):
    context = ge.get_context()
    batch_request = BatchRequest(
        datasource_name="stripe_rss",
        data_connector_name="default_runtime_data_connector_name",
        data_asset_name="stripe_blog_posts",
        runtime_parameters={"path": csv_path},
        batch_identifiers={"default_identifier_name": "validation_batch"},
    )
    checkpoint = Checkpoint(
        name="stripe_rss_checkpoint",
        context=context,
        validations=[
            {
                "batch_request": batch_request,
                "expectation_suite_name": suite_name,
            }
        ],
    )
    result = checkpoint.run()
    # Raise if any expectation failed
    if not result["success"]:
        raise RuntimeError("Data quality check failed – see validation_report.json")
    # Persist the validation report for later review
    context.save_validation_result(result)

In the dvc.yaml file the stage looks like:

stages:
  fetch_and_validate:
    cmd: python pipeline.py
    deps:
      - scripts/fetch_rss.py
      - data/stripe_blog_posts.csv
      - great_expectations/expectations/my_stripe_rss_suite.json
    outs:
      - data/stripe_blog_posts.csv
      - great_expectations/uncommitted/validations/validation_report.json

Step 5 — Adapting to Drift

If the feed adds a new column, the profiling step can be re‑run to extend the suite. I automate this with a “re‑profile” flag that only runs when the schema hash changes.

import hashlib, json, os

def schema_hash(df: pd.DataFrame) -> str:
    return hashlib.sha256(json.dumps(list(df.columns), sort_keys=True).encode()).hexdigest()

def maybe_reprofile(df, suite_name, hash_path=".schema_hash"):
    new_hash = schema_hash(df)
    old_hash = ""
    if os.path.exists(hash_path):
        with open(hash_path) as f:
            old_hash = f.read().strip()
    if new_hash != old_hash:
        create_expectation_suite(df, suite_name)
        with open(hash_path, "w") as f:
            f.write(new_hash)

Limitations and Tradeoffs

This pattern assumes the feed can be materialized as a flat CSV; deeply nested JSON would need a custom transformer. Great Expectations’ profiling is fast for < 10 k rows but becomes costly on larger batches, so you may want to sample. Finally, DVC tracks files, not directories, so you must add the whole great_expectations folder to version control to capture suite evolution.

Frequently Asked Questions

Can I use this approach with a JSON API instead of RSS?

Yes. As long as you can turn the payload into a pandas DataFrame (or a Parquet file), the same validation flow applies.

What happens if the RSS feed is temporarily down?

The fetch function raises a RuntimeError. DVC marks the stage as failed, and in CI you can add a retry policy or fall back to the last successful CSV.

Do I need to version the expectation suite itself?

Absolutely. By adding great_expectations/ to DVC you get a full history of how your quality rules evolved, which is priceless for audits.

How can I surface validation failures in a pull‑request?

Great Expectations exits with a non‑zero status when any expectation fails. In GitHub Actions or GitLab CI, simply run python pipeline.py; the job will fail and block the merge.

Is there a way to auto‑update the suite without manual re‑profiling?

You can schedule a nightly job that recomputes the schema hash and calls maybe_reprofile. If the hash changed, the suite is regenerated and committed back to the repo.

What I'd Change

If I were to rebuild this from scratch I’d replace the CSV intermediate with a Delta Lake table and let Great Expectations read directly from it. That eliminates the double‑write (CSV → validation) and gives atomic upserts when the feed adds or removes items. The trade‑off is added operational complexity, but for high‑throughput streams the performance gain is worth it.

Complete Script

The full runnable script combining all steps. Save it as pipeline.py and run python pipeline.py. It fetches the RSS, versions the data, validates it, and writes a report.

#!/usr/bin/env python3
"""
End‑to‑end pipeline:
1. Fetch Stripe Blog RSS → CSV
2. Profile & create Great Expectations suite (if schema changed)
3. Run validation checkpoint
4. Persist validation report
"""

import os
import hashlib
import json
import requests
import feedparser
import pandas as pd
import great_expectations as ge
from great_expectations.core.batch import BatchRequest
from great_expectations.checkpoint import Checkpoint

RSS_URL = "https://stripe.com/blog/feed.rss"
DATA_DIR = "data"
GE_DIR = "great_expectations"
CSV_PATH = os.path.join(DATA_DIR, "stripe_blog_posts.csv")
SUITE_NAME = "my_stripe_rss_suite"
HASH_PATH = ".schema_hash"


def fetch_rss(url: str) -> pd.DataFrame:
    """Download RSS and return a tidy DataFrame."""
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
    except requests.RequestException as e:
        raise RuntimeError(f"Failed to download RSS: {e}") from e

    feed = feedparser.parse(resp.text)
    rows = []
    for e in feed.entries:
        rows.append({
            "title": e.get("title", "").strip(),
            "link": e.get("link", ""),
            "published": e.get("published", ""),
            "author": e.get("author", ""),
            "tags": ", ".join([t.term for t in e.get("tags", [])]),
        })
    return pd.DataFrame(rows)


def save_csv(df: pd.DataFrame, path: str) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    df.to_csv(path, index=False)


def schema_hash(df: pd.DataFrame) -> str:
    """Deterministic hash of the column list."""
    return hashlib.sha256(json.dumps(list(df.columns), sort_keys=True).encode()).hexdigest()


def create_expectation_suite(df: pd.DataFrame, suite_name: str) -> None:
    """Generate a baseline suite from the current DataFrame."""
    context = ge.get_context()
    if "stripe_rss" not in context.list_datasources():
        context.add_datasource(
            "stripe_rss",
            class_name="PandasDatasource",
        )
    batch = context.get_batch({
        "datasource_name": "stripe_rss",
        "data_asset_name": "stripe_blog_posts",
        "batch_data": df,
    })
    suite = context.create_expectation_suite(suite_name, overwrite=True)
    # Generic schema checks
    batch.expect_table_row_count_to_be_between(min_value=1, max_value=1000)
    batch.expect_column_to_exist("title")
    batch.expect_column_values_to_not_be_null("title")
    batch.expect_column_to_exist("link")
    batch.expect_column_values_to_match_regex("link", r"^https?://")
    batch.expect_column_to_exist("published")
    sample_date = df["published"].iloc[0] if not df["published"].empty else ""
    if sample_date:
        batch.expect_column_values_to_match_strftime_format("published", "%a, %d %b %Y %H:%M:%S %z")
    # Business rule
    batch.expect_column_values_to_match_regex("tags", r".+")
    batch.save_expectation_suite(discard_failed_expectations=False)
    context.save_expectation_suite(suite, suite_name=suite_name)


def maybe_reprofile(df: pd.DataFrame, suite_name: str) -> None:
    """Re‑profile only when the column set changes."""
    new_hash = schema_hash(df)
    old_hash = ""
    if os.path.exists(HASH_PATH):
        with open(HASH_PATH) as f:
            old_hash = f.read().strip()
    if new_hash != old_hash:
        print("Schema change detected – rebuilding expectation suite.")
        create_expectation_suite(df, suite_name)
        with open(HASH_PATH, "w") as f:
            f.write(new_hash)
    else:
        print("Schema unchanged – reusing existing suite.")


def run_validation(csv_path: str, suite_name: str) -> None:
    """Execute a Great Expectations checkpoint; raise on failure."""
    context = ge.get_context()
    batch_request = BatchRequest(
        datasource_name="stripe_rss",
        data_connector_name="default_runtime_data_connector_name",
        data_asset_name="stripe_blog_posts",
        runtime_parameters={"path": csv_path},
        batch_identifiers={"default_identifier_name": "validation_batch"},
    )
    checkpoint = Checkpoint(
        name="stripe_rss_checkpoint",
        context=context,
        validations=[
            {
                "batch_request": batch_request,
                "expectation_suite_name": suite_name,
            }
        ],
    )
    result = checkpoint.run()
    if not result["success"]:
        raise RuntimeError("Data quality check failed – see validation_report.json")
    # Persist report for DVC tracking
    context.save_validation_result(result)


def main():
    # 1️⃣ Fetch and persist CSV
    df = fetch_rss(RSS_URL)
    save_csv(df, CSV_PATH)

    # 2️⃣ Profile / update expectation suite
    maybe_reprofile(df, SUITE_NAME)

    # 3️⃣ Validate
    try:
        run_validation(CSV_PATH, SUITE_NAME)
        print("✅ Validation passed – pipeline ready for downstream jobs.")
    except RuntimeError as ve:
        print(f"❌ Validation error: {ve}")
        raise


if __name__ == "__main__":
    main()

Expected Output

When you run python pipeline.py you should see something like:

Schema unchanged – reusing existing suite.
✅ Validation passed – pipeline ready for downstream jobs.

If the feed drops the link field, the script will abort with a clear error and DVC will mark the stage as failed.