Airflow कोड का परीक्षण

Airflow के साथ Data Pipelines बनाना

Volker Janz

Senior Developer Advocate at Astronomer

टेस्टिंग के तीन स्तर

Dag testing pyramid

Integrity: क्या Dag बिना त्रुटि के लोड होता है?
Unit: क्या बिज़नेस लॉजिक सही परिणाम देता है?
Integration: क्या पूरा Dag एंड-टू-एंड चलता है?

DagBag के साथ इंटीग्रिटी टेस्ट

from airflow.models import DagBag


dag_bag = DagBag(include_examples=False)


def test_no_import_errors():
    assert len(dag_bag.import_errors) == 0


def test_dag_loaded():
    assert "daily_etl" in dag_bag.dags

इम्पोर्ट पर Dags क्यों टूटते हैं

ModuleNotFoundError

ModuleNotFoundError: मिसिंग प्रोवाइडर पैकेज या गलत इम्पोर्ट पाथ
NameError: रीनेम किया वैरिएबल/फंक्शन या टाइपो
ImportError: Dag फाइलों में सर्कुलर इम्पोर्ट्स
इंटीग्रिटी टेस्ट इन समस्याओं से बचाते हैं

टास्क फंक्शनों का यूनिट टेस्टिंग

Dag फाइल में:

def clean_record(record):
    return {
        "name": record["name"].strip(),
        "email": record["email"].lower(),
    }

@task
def transform(records):
    return [clean_record(r)
            for r in records]

बिज़नेस लॉजिक अलग करें

टेस्ट फाइल में:

from dags.data_cleaning import (
    clean_record,
)

def test_strips_whitespace():
    result = clean_record(
      {"name": "  Alice  ",
       "email": "[email protected]"}
    )
    assert result["name"] == "Alice"

यूनिट टेस्ट बिज़नेस लॉजिक पर केंद्रित रखें

dag.test() के साथ इंटीग्रेशन टेस्ट

import pytest
from airflow.models import DagBag
from pendulum import datetime

dag_bag = DagBag(include_examples=False)


def test_etl_pipeline():
    dag = dag_bag.get_dag("etl_output")
    assert dag is not None

    dag.test(logical_date=datetime(2026, 1, 15))

    output = Path("/tmp/etl_results.json")
    assert output.exists()
    results = json.loads(output.read_text())
    assert len(results) == 2

नियंत्रित इनपुट के साथ वास्तविक Dag चलाएँ और आउटपुट वैलिडेट करें

CI में टेस्टिंग

CI pipeline

Integrity + unit: हर कमिट पर (तेज़)
Integration: पुल रिक्वेस्ट या नाइटली पर (धीमा)
सभी तीन पास किए बिना कोई Dag प्रोडक्शन तक नहीं जाता

अभ्यास करते हैं!

Airflow के साथ Data Pipelines बनाना