Data Pipeline Orchestration — Complete Guide

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

default_args = {
    "owner": "data_team",
    "depends_on_past": False,
    "email_on_failure": True,
    "email": ["alerts"@company".com"],
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

def extract_data():
    """Simulate extracting data from an API."""
    import json
    import random

    data = {
        "date": "2026-06-23",
        "records": random.randint(1000, 5000),
        "source": "api_v2",
    }
    with open("/tmp/extracted_data.json", "w") as f:
        json.dump(data, f)
    print(f"Extracted {data['records']} records")

def validate_data():
    """Validate extracted data before processing."""
    import json

    with open("/tmp/extracted_data.json") as f:
        data = json.load(f)

    assert data["records"] > 0, "No records extracted!"
    assert data["source"] in ("api_v1", "api_v2"), "Unknown source"
    print(f"Validation passed: {data['records']} records from {data['source']}")

def transform_data():
    """Transform data: filter, aggregate, enrich."""
    import json

    with open("/tmp/extracted_data.json") as f:
        data = json.load(f)

    data["status"] = "processed"
    data["processing_time"] = datetime.now().isoformat()

    with open("/tmp/transformed_data.json", "w") as f:
        json.dump(data, f)
    print(f"Transform complete for {data['date']}")

with DAG(
    dag_id="simple_etl_pipeline",
    default_args=default_args,
    description="A simple ETL pipeline example",
    schedule_interval="0 6 * * *",
    start_date=days_ago(1),
    catchup=False,
    tags=["etl"],
) as dag:

    extract = PythonOperator(
        task_id="extract_data",
        python_callable=extract_data,
    )

    validate = PythonOperator(
        task_id="validate_data",
        python_callable=validate_data,
    )

    transform = PythonOperator(
        task_id="transform_data",
        python_callable=transform_data,
    )

    notify = BashOperator(
        task_id="notify_completion",
        bash_command='echo "Pipeline completed for {{ ds }}"',
    )

    extract >> validate >> transform >> notify

from datetime import timedelta
from airflow import DAG
from airflow.operators.python import BranchPythonOperator, PythonOperator
from airflow.utils.dates import days_ago
import json
import random

def check_data_quality(**context):
    """Branch based on data quality check."""
    with open("/tmp/extracted_data.json") as f:
        data = json.load(f)

    quality_score = random.uniform(0.8, 1.0)
    context["ti"].xcom_push(key="quality_score", value=quality_score)

    if quality_score >= 0.95:
        return "process_data"
    elif quality_score >= 0.85:
        return "flag_for_review"
    else:
        return "send_alert"

def process_data():
    print("Processing high-quality data...")

def flag_for_review():
    print("Flagging data for manual review...")

def send_alert():
    print("Sending alert: data quality below threshold...")

with DAG(
    dag_id="quality_gated_pipeline",
    schedule_interval="@daily",
    start_date=days_ago(1),
    catchup=False,
    tags=["quality"],
) as dag:

    check_quality = BranchPythonOperator(
        task_id="check_data_quality",
        python_callable=check_data_quality,
    )

    process = PythonOperator(
        task_id="process_data",
        python_callable=process_data,
    )

    review = PythonOperator(
        task_id="flag_for_review",
        python_callable=flag_for_review,
    )

    alert = PythonOperator(
        task_id="send_alert",
        python_callable=send_alert,
    )

    check_quality >> [process, review, alert]

import time
import random

def unreliable_task():
    """Simulate an API call that occasionally fails."""
    attempt = random.randint(1, 3)
    print(f"Attempt {attempt} of 3")

    if attempt < 3:
        # Simulate a transient error
        raise ConnectionError("API timeout, retrying...")

    print("API call succeeded on attempt 3")

def retry_simulation():
    """Simulate Airflow's retry mechanism."""
    max_retries = 2
    retry_delay = 1

    for attempt in range(1, max_retries + 2):
        try:
            print(f"\n=== Task attempt {attempt} ===")
            if attempt < 3:
                raise ConnectionError(f"Simulated failure on attempt {attempt}")
            print("Task completed successfully!")
            return
        except ConnectionError as e:
            print(f"Failed: {e}")
            if attempt <= max_retries:
                print(f"Retrying in {retry_delay}s...")
                time.sleep(retry_delay)
            else:
                print("Max retries exceeded. Marking task as failed.")

retry_simulation()

import time
from datetime import datetime

def simulate_file_sensor():
    """Simulate Airflow's FileSensor."""
    files_to_check = [
        "/data/orders/2026-06-23/part-00001.parquet",
        "/data/orders/2026-06-23/part-00002.parquet",
        "/data/orders/2026-06-23/part-00003.parquet",
    ]

    available_files = set()
    timeout = 10
    start = time.time()

    while len(available_files) < len(files_to_check):
        elapsed = time.time() - start
        if elapsed > timeout:
            print(f"TIMEOUT: Only {len(available_files)}/{len(files_to_check)} files found")
            return False

        for f in files_to_check:
            if f not in available_files:
                if f.endswith("part-00002.parquet"):
                    continue
                available_files.add(f)
                print(f"[{datetime.now().strftime('%H:%M:%S')}] Found: {f}")

        time.sleep(1)

    print(f"All {len(available_files)} files available. Proceeding...")
    return True

simulate_file_sensor()