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

Reproducible Pipelines at Scale: Versioning Data Artifacts with DVC in Production

Mausam

Ever pulled a JSON feed in a DAG, only to discover that a downstream model broke because the upstream API added a field you never handled? I hit that wall twice while extending the RSS ingestion pipeline from our Airflow deep‑dive, and the fix wasn’t “add a try/except” but “freeze the exact input I used to ship the model”. In this post you’ll learn, step by step, how to bring a live API under DVC control, turn raw data into a reproducible processing stage, and safely roll back when the source drifts. By the end you’ll have a production‑ready script that you can drop into any Airflow task and trust that the same results will be produced tomorrow, next week, or after a teammate pushes a change.

Key Takeaways

  • DVC can version even tiny JSON payloads, giving you a Git‑backed audit trail for every ingestion run.
  • Parameterizing transformations and committing the derived artifacts lets you rebuild any pipeline state from a single commit.
  • When source data drifts, dvc checkout restores the exact version you used for a historic model, eliminating silent regressions.
  • Remote storage (a simple local folder in this demo) scales to multi‑GB artifacts without bloating your Git repo.
  • Integrating DVC commands in Airflow tasks is straightforward: treat them as BashOperators or invoke the Python API.

The Problem

Our Airflow‑driven RSS ingestion runs every hour, fetching XML, converting it to Parquet, and loading it into BigQuery. When we added a new downstream ML model that consumes the same raw feed, we realized that a single change in the upstream schema broke the model overnight. The root cause was that the raw feed wasn’t versioned—Git only tracked code, not the data that fed the model. Without a deterministic snapshot, debugging became a guessing game.

Data and Sources

We’ll use the public JSONPlaceholder Todos endpoint as a stand‑in for any live JSON API:

Data accessed on 2026-11-02.

Step 1 — Ingesting and Initial Data Versioning from a Live API

Sub‑problem: How do we reliably fetch external data and lock it down before any transformation?

import os, json, requests, subprocess, sys

RAW_DIR = "data/raw"
RAW_FILE = os.path.join(RAW_DIR, "todos.json")

def fetch_and_version():
    os.makedirs(RAW_DIR, exist_ok=True)
    try:
        resp = requests.get("https://jsonplaceholder.typicode.com/todos", timeout=10)
        resp.raise_for_status()
    except requests.RequestException as e:
        print(f"❌ Failed to fetch API: {e}")
        sys.exit(1)

    with open(RAW_FILE, "w") as f:
        json.dump(resp.json(), f, indent=2)

    # Add the file to DVC and commit the .dvc metadata
    try:
        subprocess.run(["dvc", "add", RAW_FILE], check=True, capture_output=True)
    except subprocess.CalledProcessError as e:
        print(f"❌ DVC add failed: {e.stderr.decode().strip()}")
        sys.exit(1)

    # Git commit the .dvc file (assumes you are inside a git repo)
    subprocess.run(["git", "add", f"{RAW_FILE}.dvc"], check=True)
    subprocess.run(["git", "commit", "-m", "Add raw todos.json with DVC"], check=True)

The snippet does three things: (1) fetches the JSON payload, (2) writes it to a stable location, and (3) runs dvc add so the file’s checksum lives in .dvc. Because the .dvc file is tracked by Git, any clone of the repo can reconstruct the exact raw data used in that commit.

Step 2 — Building a Parameterized Processing Pipeline Stage

Sub‑problem: How do we apply deterministic transformations and capture their dependencies?

import pandas as pd
import matplotlib.pyplot as plt

PROC_DIR = "data/processed"
PROC_FILE = os.path.join(PROC_DIR, "todos_by_user.csv")
PLOT_FILE = "output.png"

# Example parameter: keep only completed tasks
PARAMS = {"keep_completed": True}

def process_and_version():
    os.makedirs(PROC_DIR, exist_ok=True)

    # Load raw JSON via DVC’s Python API (fallback to local file)
    try:
        import dvc.api
        with dvc.api.open(RAW_FILE, repo=".", rev="HEAD") as f:
            raw = json.load(f)
    except Exception:
        # If DVC API fails, read the file directly (useful for first run)
        with open(RAW_FILE) as f:
            raw = json.load(f)

    df = pd.DataFrame(raw)

    # Apply parameterized filter
    if PARAMS["keep_completed"]:
        df = df[df["completed"]]

    # Aggregate: count tasks per user
    agg = df.groupby("userId").size().reset_index(name="task_count")
    agg.to_csv(PROC_FILE, index=False)

    # Version the processed CSV
    subprocess.run(["dvc", "add", PROC_FILE], check=True, capture_output=True)
    subprocess.run(["git", "add", f"{PROC_FILE}.dvc"], check=True)
    subprocess.run(["git", "commit", "-m", "Add processed todos_by_user.csv"], check=True)

    # Plot for quick visual sanity check
    agg.plot(kind="bar", x="userId", y="task_count", legend=False)
    plt.title("Completed Todos per User")
    plt.xlabel("User ID")
    plt.ylabel("Count")
    plt.tight_layout()
    plt.savefig(PLOT_FILE)

We read the raw artifact through dvc.api.open so the exact version is enforced. The transformation is fully driven by the PARAMS dict; changing a parameter forces a new run and a new versioned CSV.

Step 3 — Simulating Data Drift and Ensuring Reproducibility

Sub‑problem: How do we handle a changed source without losing the ability to reproduce old results?

def simulate_drift_and_revert():
    # Simulate drift by re‑fetching the API (the placeholder API is static,
    # so we artificially modify the file to emulate a change)
    with open(RAW_FILE, "r+") as f:
        data = json.load(f)
        # Flip the 'completed' flag of the first record
        data[0]["completed"] = not data[0]["completed"]
        f.seek(0)
        json.dump(data, f, indent=2)
        f.truncate()

    # Re‑add to DVC – this creates a new version
    subprocess.run(["dvc", "add", RAW_FILE], check=True, capture_output=True)
    subprocess.run(["git", "add", f"{RAW_FILE}.dvc"], check=True)
    subprocess.run(["git", "commit", "-m", "Simulate source drift"], check=True)

    # Now revert to the previous version
    # HEAD~1 points to the commit before the drift
    subprocess.run(["git", "checkout", "HEAD~1"], check=True)
    subprocess.run(["dvc", "checkout"], check=True)  # restores original raw file
    print("🔁 Reverted to original raw data version.")

After the artificial drift we commit a new DVC version, then use git checkout + dvc checkout to roll back. This pattern works in production: a nightly Airflow task can detect a checksum mismatch, abort, and automatically revert to the last known‑good version.

Step 4 — Collaborative Workflows and Remote Data Storage

Sub‑problem: How do teams share large versioned artifacts without blowing up the Git repo?

def configure_remote_and_sync():
    # Remote is a local folder for demo; replace with S3, GCS, etc. in prod
    remote_path = os.path.abspath("dvc_remote")
    os.makedirs(remote_path, exist_ok=True)

    # Add remote if not already configured
    subprocess.run(["dvc", "remote", "add", "-d", "local", remote_path], check=True)
    # Push all tracked artifacts
    subprocess.run(["dvc", "push"], check=True)
    print(f"🚀 Pushed data to remote at {remote_path}")

def pull_from_remote():
    subprocess.run(["dvc", "pull"], check=True)
    print("🔽 Pulled latest data from remote.")

In a real team you’d point remote_path to an object store (S3, GCS, Azure). The push/pull commands are idempotent; if the remote is temporarily unavailable, DVC falls back to the local cache, and the Airflow task can still run with the cached version.

Limitations and Tradeoffs

This approach shines when you have repeatable, versioned artifacts that are larger than a few kilobytes. For tiny JSON files the overhead of a DVC checkout may feel heavy, and a simple Git commit could suffice. DVC also introduces an extra dependency and a learning curve for CI/CD pipelines. Finally, remote storage costs and latency become a factor at petabyte scale; you’ll need to tune cache size and remote bandwidth.

Frequently Asked Questions

Do I have to commit the .dvc files to Git?

Yes. The .dvc files store the checksum and path metadata. Without them Git cannot point to a specific data version.

Can I run DVC commands inside an Airflow BashOperator?

Absolutely. Treat dvc add, dvc push, and dvc checkout as any other CLI tool. If you prefer Python, dvc.api offers read‑only access to versioned files.

What happens if the remote storage is down?

DVC falls back to the local cache, so your DAG can still read already‑cached artifacts. New pushes will fail, but you can retry later.

Is DVC overkill for CSVs under 1 MB?

For very small files the performance gain is marginal, but the pattern scales. If you anticipate growth or need strict reproducibility, start with DVC now rather than refactor later.

How do I version code together with data?

Keep your pipeline scripts in the same Git repo as the .dvc files. A single commit then captures both the code and the exact data version it expects.

What I'd Change

If I were to rebuild this pipeline for a high‑throughput production environment, I’d replace the local‑folder remote with an S3 bucket and enable DVC’s --jobs flag to parallelize pushes of large artifacts. I’d also wrap every DVC CLI call in a reusable Python helper that retries on transient network errors, turning the brittle subprocess calls into a robust library. Finally, I’d integrate a checksum validation step in the Airflow DAG itself so that a mismatch aborts the run before any downstream model is touched.

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
Reproducible pipeline demo using DVC and the JSONPlaceholder Todos API.
Run with: python dvc_pipeline_demo.py
"""

import os
import sys
import json
import subprocess
import requests
import pandas as pd
import matplotlib.pyplot as plt

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
RAW_DIR = "data/raw"
PROC_DIR = "data/processed"
RAW_FILE = os.path.join(RAW_DIR, "todos.json")
PROC_FILE = os.path.join(PROC_DIR, "todos_by_user.csv")
PLOT_FILE = "output.png"
REMOTE_NAME = "local"
REMOTE_PATH = os.path.abspath("dvc_remote")

# Parameter dict – change to trigger a new version
PARAMS = {"keep_completed": True}


# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
def run_cmd(cmd, err_msg):
    """Run a shell command, raise on failure with a friendly message."""
    try:
        result = subprocess.run(
            cmd, check=True, capture_output=True, text=True
        )
        return result.stdout.strip()
    except subprocess.CalledProcessError as e:
        print(f"❌ {err_msg}: {e.stderr.strip()}")
        sys.exit(1)


def fetch_and_version():
    """Step 1 – Pull live JSON and version it with DVC."""
    os.makedirs(RAW_DIR, exist_ok=True)
    try:
        resp = requests.get(
            "https://jsonplaceholder.typicode.com/todos", timeout=10
        )
        resp.raise_for_status()
    except requests.RequestException as e:
        print(f"❌ Failed to fetch API: {e}")
        sys.exit(1)

    with open(RAW_FILE, "w") as f:
        json.dump(resp.json(), f, indent=2)

    run_cmd(["dvc", "add", RAW_FILE], "DVC add failed")
    run_cmd(["git", "add", f"{RAW_FILE}.dvc"], "Git add failed")
    run_cmd(
        ["git", "commit", "-m", "Add raw todos.json with DVC"],
        "Git commit failed",
    )
    print("✅ Raw data fetched and versioned.")


def process_and_version():
    """Step 2 – Transform raw data, version the result, and plot."""
    os.makedirs(PROC_DIR, exist_ok=True)

    # Load raw via DVC API if possible; fallback to file
    try:
        import dvc.api

        with dvc.api.open(RAW_FILE, repo=".", rev="HEAD") as f:
            raw = json.load(f)
    except Exception:
        with open(RAW_FILE) as f:
            raw = json.load(f)

    df = pd.DataFrame(raw)

    if PARAMS["keep_completed"]:
        df = df[df["completed"]]

    agg = df.groupby("userId").size().reset_index(name="task_count")
    agg.to_csv(PROC_FILE, index=False)

    run_cmd(["dvc", "add", PROC_FILE], "DVC add processed failed")
    run_cmd(["git", "add", f"{PROC_FILE}.dvc"], "Git add processed failed")
    run_cmd(
        ["git", "commit", "-m", "Add processed todos_by_user.csv"],
        "Git commit processed failed",
    )
    print("✅ Processed data versioned.")

    # Plotting
    agg.plot(kind="bar", x="userId", y="task_count", legend=False)
    plt.title("Completed Todos per User")
    plt.xlabel("User ID")
    plt.ylabel("Count")
    plt.tight_layout()
    plt.savefig(PLOT_FILE)
    print(f"📈 Plot saved to {PLOT_FILE}")


def simulate_drift_and_revert():
    """Step 3 – Emulate a source change and roll back."""
    # Artificial drift: flip a flag
    with open(RAW_FILE, "r+") as f:
        data = json.load(f)
        data[0]["completed"] = not data[0]["completed"]
        f.seek(0)
        json.dump(data, f, indent=2)
        f.truncate()

    run_cmd(["dvc", "add", RAW_FILE], "DVC add after drift failed")
    run_cmd(["git", "add", f"{RAW_FILE}.dvc"], "Git add after drift failed")
    run_cmd(
        ["git", "commit", "-m", "Simulate source drift"], "Git commit drift failed"
    )
    print("⚠️ Simulated data drift and created new DVC version.")

    # Revert to previous commit
    run_cmd(["git", "checkout", "HEAD~1"], "Git checkout failed")
    run_cmd(["dvc", "checkout"], "DVC checkout failed")
    print("🔁 Reverted to original raw data version.")


def configure_remote_and_sync():
    """Step 4 – Set up a local remote and push artifacts."""
    os.makedirs(REMOTE_PATH, exist_ok=True)
    # Add remote only if not already present
    existing = run_cmd(
        ["dvc", "remote", "list"], "Failed to list DVC remotes"
    )
    if REMOTE_NAME not in existing:
        run_cmd(
            ["dvc", "remote", "add", "-d", REMOTE_NAME, REMOTE_PATH],
            "Failed to add remote",
        )
    run_cmd(["dvc", "push"], "DVC push failed")
    print(f"🚀 Data pushed to remote at {REMOTE_PATH}")


def pull_from_remote():
    """Pull latest data from remote (useful on a fresh clone)."""
    run_cmd(["dvc", "pull"], "DVC pull failed")
    print("🔽 Pulled data from remote.")


def main():
    # Fresh start: ensure remote is configured
    configure_remote_and_sync()

    # 1️⃣ Ingest & version raw API data
    fetch_and_version()

    # 2️⃣ Transform, version processed data, and plot
    process_and_version()

    # 3️⃣ Simulate drift & demonstrate reproducibility
    simulate_drift_and_revert()

    # 4️⃣ Pull to verify remote sync (optional)
    pull_from_remote()


if __name__ == "__main__":
    main()

Expected Output

Running python dvc_pipeline_demo.py should print a series of status messages ending with:

✅ Raw data fetched and versioned.
✅ Processed data versioned.
📈 Plot saved to output.png
⚠️ Simulated data drift and created new DVC version.
🔁 Reverted to original raw data version.
🚀 Data pushed to remote at /full/path/to/dvc_remote
🔽 Pulled data from remote.

You’ll also see an output.png bar chart showing the count of completed todos per user.

إرسال تعليق

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