Data Orchestration & Workflow Management

Coordinating complex data pipelines with Apache Airflow, Prefect, and modern orchestration tools

The Backbone of Production Data Pipelines

Data orchestration is the automated coordination of complex workflows that move and transform data across systems. Think of it as the conductor of an orchestra, ensuring each instrument (task) plays at the right time, in the right order, handling failures gracefully. Without orchestration, production data pipelines become fragile scripts held together with cron jobs and hope. Modern orchestration tools like Apache Airflow, Prefect, and Dagster provide scheduling, dependency management, monitoring, and error handling essential for reliable data operations.

What is Data Orchestration?

Data orchestration automates the execution of data workflows, ensuring tasks run in the correct sequence, dependencies are respected, and failures are handled appropriately.

WITHOUT ORCHESTRATION:
┌─────────────────────────────────────────────────────┐
│  Cron: 0 2 * * * python extract.py                  │
│  Cron: 0 3 * * * python transform.py  ← Hope        │
│  Cron: 0 4 * * * python load.py       extract       │
│                                        finished!    │
└─────────────────────────────────────────────────────┘
Problems: No dependency tracking, no retry logic,
          no visibility, manual failure handling


WITH ORCHESTRATION:
┌──────────────────────────────────────────┐
│            Orchestrator                  │
│  ┌──────────┐                            │
│  │ extract  │──success──┐                │
│  └──────────┘           ▼                │
│                  ┌─────────────┐         │
│                  │ transform   │─success─┤
│                  └─────────────┘         │
│                         │                │
│                    ┌────┴────┐           │
│                    ▼         ▼           │
│              ┌──────┐   ┌──────┐         │
│              │ load │   │alert │         │
│              └──────┘   └──────┘         │
└──────────────────────────────────────────┘
Benefits: Dependencies enforced, retries automatic,
          full visibility, smart failure handling

Orchestration coordinates task execution with dependency tracking and monitoring

Core Capabilities

📅 Scheduling

Run workflows on time-based triggers (cron), event-based triggers, or manually

🔗 Dependencies

Define task relationships, ensure correct execution order, handle complex graphs

🔄 Retries & Recovery

Automatic retries with exponential backoff, graceful failure handling

📊 Monitoring

Real-time visibility into pipeline execution, logs, and metrics

🚨 Alerting

Notify teams via email, Slack, PagerDuty when failures or SLA breaches occur

⚖️ Resource Management

Control parallelism, manage compute resources, prioritize critical workloads

Apache Airflow: The Industry Standard

Apache Airflow is the most widely adopted open-source workflow orchestration platform, originally developed by Airbnb. It uses Python to define workflows as Directed Acyclic Graphs (DAGs) with rich operators for common tasks.

Core Concepts

DAG (Directed Acyclic Graph)

A collection of tasks with defined dependencies and execution order

         A (extract)
         │
         ▼
         B (transform)
         │
    ┌────┴────┐
    ▼         ▼
    C         D  (load to different targets)
   (DWH)    (Lake)
    │         │
    └────┬────┘
         ▼
         E (alert)

Directed: Flows in one direction (no loops)
Acyclic: No circular dependencies allowed

Tasks execute in topological order respecting dependencies

Example: Complete Airflow DAG

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta
import pandas as pd

# Define default arguments for all tasks
default_args = {
    'owner': 'data-team',
    'depends_on_past': False,      # Don't wait for previous runs
    'email': ['alerts@company.com'],
    'email_on_failure': True,      # Alert on failures
    'email_on_retry': False,
    'retries': 3,                  # Retry failed tasks 3 times
    'retry_delay': timedelta(minutes=5),  # Wait 5 min between retries
    'retry_exponential_backoff': True,    # Exponential backoff
    'max_retry_delay': timedelta(minutes=30),
}

# Define the DAG
dag = DAG(
    'sales_data_pipeline',
    default_args=default_args,
    description='Daily sales data ETL pipeline',
    schedule='0 2 * * *',  # Run at 2 AM daily (cron syntax)
    start_date=datetime(2024, 1, 1),
    catchup=False,                  # Don't backfill historical runs
    tags=['sales', 'daily', 'production'],
)

# Task 1: Extract sales data
def extract_sales(**context):
    """Extract yesterday's sales from source database"""
    from sqlalchemy import create_engine

    engine = create_engine('postgresql://source_db')

    # Extract with execution date context
    execution_date = context['ds']  # YYYY-MM-DD format

    query = f"""
        SELECT * FROM sales
        WHERE date = '{execution_date}'
    """

    df = pd.read_sql(query, engine)

    # Save to temp location
    output_path = f'/tmp/sales_{execution_date}.parquet'
    df.to_parquet(output_path)

    # Push path to XCom for downstream tasks
    context['task_instance'].xcom_push(key='sales_file', value=output_path)

    print(f"Extracted {len(df)} rows for {execution_date}")

extract_task = PythonOperator(
    task_id='extract_sales',
    python_callable=extract_sales,
    dag=dag,
)

# Task 2: Validate data quality
def validate_data(**context):
    """Run data quality checks"""
    # Pull file path from previous task
    ti = context['task_instance']
    sales_file = ti.xcom_pull(task_ids='extract_sales', key='sales_file')

    df = pd.read_parquet(sales_file)

    # Data quality checks
    assert df['customer_id'].notnull().all(), "Null customer_ids found!"
    assert df['amount'].min() >= 0, "Negative amounts found!"
    assert len(df) > 0, "No data extracted!"

    print(f"✓ Data quality checks passed: {len(df)} rows validated")

validate_task = PythonOperator(
    task_id='validate_data',
    python_callable=validate_data,
    dag=dag,
)

# Task 3: Transform data
def transform_sales(**context):
    """Clean and aggregate sales data"""
    ti = context['task_instance']
    sales_file = ti.xcom_pull(task_ids='extract_sales', key='sales_file')

    df = pd.read_parquet(sales_file)

    # Transformations
    df = df.drop_duplicates()
    df['amount'] = df['amount'].abs()

    # Aggregate by customer
    customer_summary = df.groupby('customer_id').agg({
        'amount': 'sum',
        'order_id': 'count'
    }).reset_index()
    customer_summary.columns = ['customer_id', 'total_sales', 'order_count']

    # Save transformed data
    output_path = sales_file.replace('.parquet', '_transformed.parquet')
    customer_summary.to_parquet(output_path)
    ti.xcom_push(key='transformed_file', value=output_path)

    print(f"Transformed data: {len(customer_summary)} customers")

transform_task = PythonOperator(
    task_id='transform_sales',
    python_callable=transform_sales,
    dag=dag,
)

# Task 4: Load to data warehouse
load_task = PostgresOperator(
    task_id='load_to_warehouse',
    postgres_conn_id='warehouse_db',
    sql="""
        INSERT INTO sales_summary (customer_id, total_sales, order_count, date)
        SELECT customer_id, total_sales, order_count, '{{ ds }}' as date
        FROM staging.customer_summary
        ON CONFLICT (customer_id, date)
        DO UPDATE SET
            total_sales = EXCLUDED.total_sales,
            order_count = EXCLUDED.order_count
    """,
    dag=dag,
)

# Task 5: Send success notification
notify_task = BashOperator(
    task_id='send_notification',
    bash_command='curl -X POST https://hooks.slack.com/... -d "{\"text\":\"Sales pipeline completed for {{ ds }}\"}"',
    dag=dag,
)

# Define task dependencies
extract_task >> validate_task >> transform_task >> load_task >> notify_task

Result:
DAG runs daily at 2 AM
Tasks execute in sequence: Extract → Validate → Transform → Load → Notify
Failed tasks retry 3 times with 5-minute delays
Email alerts sent on failures, Slack notification on success

Airflow Architecture

┌──────────────────────────────────────────────────────┐
│                  Airflow Webserver                   │
│              (UI, API, User Interface)               │
└───────────────────┬──────────────────────────────────┘
                    │
┌───────────────────▼──────────────────────────────────┐
│                  Airflow Scheduler                   │
│   (Reads DAGs, schedules tasks, monitors state)      │
└───────────────────┬──────────────────────────────────┘
                    │
                    ▼
        ┌───────────────────────┐
        │  Metadata Database    │
        │  (PostgreSQL/MySQL)   │
        │  Stores: DAG runs,    │
        │  task states, logs    │
        └───────────┬───────────┘
                    │
        ┌───────────▼───────────┐
        │    Message Broker     │
        │  (Redis/RabbitMQ)     │
        │  Task queue           │
        └───────────┬───────────┘
                    │
        ┌───────────▼───────────────┐
        │      Airflow Workers      │
        │   (Execute actual tasks)  │
        │   Can scale horizontally  │
        └───────────────────────────┘

Distributed architecture allows scaling workers independently

Airflow Operators

Airflow provides 100+ operators for common tasks, avoiding boilerplate code.

PythonOperator

Execute Python functions

PythonOperator(python_callable=func)

BashOperator

Run bash commands

BashOperator(bash_command='...')

PostgresOperator

Execute SQL on PostgreSQL

PostgresOperator(sql='...')

S3Operator

Interact with AWS S3

S3ToRedshiftOperator(...)

HttpOperator

Make HTTP/API calls

HttpOperator(endpoint='...')

SparkSubmitOperator

Submit Spark jobs

SparkSubmitOperator(...)

Pros & Cons

✅ Pros

Industry standard, huge community
Rich ecosystem (100+ operators)
Powerful Python-based DAGs
Excellent monitoring UI
Mature, battle-tested
Managed options (MWAA, Cloud Composer)

❌ Cons

Steep learning curve
Complex architecture (scheduler, workers, DB)
DAG writing can be verbose
Scheduler can struggle at massive scale
Local development setup is heavy
Dynamic DAGs are tricky

Modern Alternatives: Prefect & Dagster

Newer orchestration tools address Airflow's pain points with better developer experience, improved observability, and modern architecture patterns.

Prefect: Negative Engineering

Developer-friendly orchestration with dynamic workflows

Prefect focuses on developer experience and dynamic workflows. Instead of static DAGs, you write normal Python code that Prefect tracks and orchestrates.

Example: Prefect Flow

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta
import pandas as pd

# Define tasks with decorators
@task(retries=3, retry_delay_seconds=300)
def extract_sales(date: str) -> pd.DataFrame:
    """Extract sales data"""
    print(f"Extracting sales for {date}")

    from sqlalchemy import create_engine
    engine = create_engine('postgresql://source_db')

    df = pd.read_sql(
        f"SELECT * FROM sales WHERE date = '{date}'",
        engine
    )

    return df

@task
def validate_data(df: pd.DataFrame) -> pd.DataFrame:
    """Validate data quality"""
    assert df['customer_id'].notnull().all(), "Null customer IDs!"
    assert df['amount'].min() >= 0, "Negative amounts!"
    assert len(df) > 0, "No data!"

    print(f"✓ Validated {len(df)} rows")
    return df

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def transform_sales(df: pd.DataFrame) -> pd.DataFrame:
    """Transform and aggregate sales"""
    df = df.drop_duplicates()
    df['amount'] = df['amount'].abs()

    summary = df.groupby('customer_id').agg({
        'amount': 'sum',
        'order_id': 'count'
    }).reset_index()
    summary.columns = ['customer_id', 'total_sales', 'order_count']

    print(f"Aggregated to {len(summary)} customers")
    return summary

@task
def load_to_warehouse(df: pd.DataFrame, date: str):
    """Load data to warehouse"""
    from sqlalchemy import create_engine
    engine = create_engine('postgresql://warehouse')

    df['date'] = date
    df.to_sql('sales_summary', engine, if_exists='append', index=False)

    print(f"Loaded {len(df)} rows to warehouse")

# Define the flow (no need for complex DAG setup!)
@flow(name="sales-pipeline")
def sales_data_pipeline(date: str):
    """Main sales pipeline flow"""

    # Extract
    raw_data = extract_sales(date)

    # Validate
    validated_data = validate_data(raw_data)

    # Transform
    transformed_data = transform_sales(validated_data)

    # Load
    load_to_warehouse(transformed_data, date)

    print(f"✓ Pipeline completed for {date}")

# Run the flow
if __name__ == "__main__":
    from datetime import datetime

    sales_data_pipeline(
        date=datetime.now().strftime("%Y-%m-%d")
    )

# Deploy with schedule
from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule

deployment = Deployment.build_from_flow(
    flow=sales_data_pipeline,
    name="daily-sales-pipeline",
    schedule=CronSchedule(cron="0 2 * * *"),  # 2 AM daily
    parameters={"date": "{{ execution_date }}"}
)

Result:
Clean, Pythonic code with automatic dependency tracking
Tasks retry 3 times with 5-minute delays
Transformed data cached for 1 hour (saves computation)
Deploy once, run anywhere (cloud or local)

Key Features

Dynamic Workflows

Write normal Python code, dependencies tracked automatically

Hybrid Execution

Run anywhere: local, cloud, Kubernetes, serverless

Caching

Smart task result caching to avoid redundant computation

Negative Engineering

Philosophy: Make failures visible but don't block execution unnecessarily

Dagster: Data-Aware Orchestration

Asset-based approach with built-in testing and data quality

Dagster takes a fundamentally different approach: instead of task-centric workflows, it focuses on data assets. You define what data you want to produce, and Dagster figures out how to build it.

Example: Dagster Pipeline

from dagster import (
    asset, AssetExecutionContext,
    Definitions, ScheduleDefinition,
    AssetCheckResult, asset_check
)
import pandas as pd

# Define data assets (not tasks!)
@asset(
    group_name="sales",
    description="Raw sales data extracted from source database"
)
def raw_sales(context: AssetExecutionContext) -> pd.DataFrame:
    """Extract raw sales data"""
    from sqlalchemy import create_engine

    date = context.partition_key  # Date partition
    engine = create_engine('postgresql://source_db')

    df = pd.read_sql(
        f"SELECT * FROM sales WHERE date = '{date}'",
        engine
    )

    context.log.info(f"Extracted {len(df)} rows for {date}")
    return df

# Asset that depends on raw_sales
@asset(
    group_name="sales",
    description="Cleaned and validated sales data"
)
def validated_sales(context: AssetExecutionContext, raw_sales: pd.DataFrame) -> pd.DataFrame:
    """Validate and clean sales data"""

    # Data quality checks
    initial_count = len(raw_sales)

    # Remove nulls
    df = raw_sales.dropna(subset=['customer_id'])

    # Fix negative amounts
    df['amount'] = df['amount'].abs()

    # Filter invalid
    df = df[df['amount'] > 0]

    context.log.info(f"Cleaned: {initial_count} → {len(df)} rows")
    return df

# Define asset checks (data quality tests)
@asset_check(asset=validated_sales)
def check_sales_quality(validated_sales: pd.DataFrame) -> AssetCheckResult:
    """Ensure sales data meets quality standards"""

    # Check for nulls
    null_count = validated_sales['customer_id'].isnull().sum()
    if null_count > 0:
        return AssetCheckResult(
            passed=False,
            description=f"Found {null_count} null customer IDs"
        )

    # Check for negative amounts
    neg_count = (validated_sales['amount'] < 0).sum()
    if neg_count > 0:
        return AssetCheckResult(
            passed=False,
            description=f"Found {neg_count} negative amounts"
        )

    # Check minimum row count
    if len(validated_sales) < 100:
        return AssetCheckResult(
            passed=False,
            description=f"Only {len(validated_sales)} rows (expected >100)"
        )

    return AssetCheckResult(
        passed=True,
        description=f"All checks passed for {len(validated_sales)} rows"
    )

@asset(
    group_name="sales",
    description="Customer sales summary for analytics"
)
def customer_sales_summary(validated_sales: pd.DataFrame) -> pd.DataFrame:
    """Aggregate sales by customer"""

    summary = validated_sales.groupby('customer_id').agg({
        'amount': 'sum',
        'order_id': 'count'
    }).reset_index()
    summary.columns = ['customer_id', 'total_sales', 'order_count']

    return summary

@asset(
    group_name="sales",
    description="Sales summary loaded to data warehouse"
)
def sales_summary_warehouse(
    context: AssetExecutionContext,
    customer_sales_summary: pd.DataFrame
) -> None:
    """Load summary to warehouse"""
    from sqlalchemy import create_engine

    engine = create_engine('postgresql://warehouse')

    customer_sales_summary['date'] = context.partition_key
    customer_sales_summary.to_sql(
        'sales_summary',
        engine,
        if_exists='append',
        index=False
    )

    context.log.info(f"Loaded {len(customer_sales_summary)} rows to warehouse")

# Define schedule
daily_schedule = ScheduleDefinition(
    name="daily_sales_pipeline",
    cron_schedule="0 2 * * *",  # 2 AM daily
    target="*",  # All assets
)

# Package everything
defs = Definitions(
    assets=[raw_sales, validated_sales, customer_sales_summary, sales_summary_warehouse],
    asset_checks=[check_sales_quality],
    schedules=[daily_schedule],
)

Result:
Asset lineage automatically tracked (raw → validated → summary → warehouse)
Data quality checks run automatically before downstream assets
If check_sales_quality fails, customer_sales_summary won't run
Beautiful UI shows asset dependencies and data flow

Key Features

Asset-Based

Focus on data you want (assets), not tasks. Clearer mental model

Built-in Testing

Asset checks run automatically, first-class data quality support

Type System

Strong typing catches errors before runtime

Data Lineage

Automatic tracking of data dependencies and asset relationships

Comparison: Airflow vs Prefect vs Dagster

Feature	Airflow	Prefect	Dagster
Paradigm	Task-based DAGs	Dynamic flows	Asset-based
Learning Curve	Steep	Gentle	Moderate
Developer Experience	Good	Excellent	Excellent
Dynamic Workflows	Limited	Native support	Good
Data Quality	Manual	Manual	Built-in (asset checks)
Testing	Requires setup	Easy	Excellent (unit testable)
Local Development	Heavy setup	Lightweight	Lightweight
Community	Huge	Growing fast	Growing
Best For	Complex enterprise pipelines	Dynamic workflows, rapid development	Data teams focused on quality

Cloud Orchestration Services

Cloud providers offer managed orchestration services that integrate seamlessly with their ecosystems, reducing operational overhead.

AWS Step Functions

Serverless workflow orchestration for AWS services

AWS Step Functions orchestrates AWS services (Lambda, Glue, ECS, etc.) using visual workflows defined in JSON (Amazon States Language). Fully managed, serverless, and deeply integrated with AWS.

Example: Step Functions Workflow

# Define workflow in Amazon States Language (JSON)
{
  "Comment": "Sales ETL Pipeline",
  "StartAt": "ExtractSales",
  "States": {
    "ExtractSales": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:extract-sales",
      "ResultPath": "$.extractResult",
      "Next": "ValidateData",
      "Retry": [
        {
          "ErrorEquals": ["States.ALL"],
          "IntervalSeconds": 300,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "NotifyFailure"
        }
      ]
    },
    "ValidateData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:validate-data",
      "ResultPath": "$.validateResult",
      "Next": "TransformData"
    },
    "TransformData": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun.sync",
      "Parameters": {
        "JobName": "transform-sales-job",
        "Arguments": {
          "--input_path.$": "$.extractResult.outputPath"
        }
      },
      "ResultPath": "$.transformResult",
      "Next": "LoadToWarehouse"
    },
    "LoadToWarehouse": {
      "Type": "Task",
      "Resource": "arn:aws:states:::ecs:runTask.sync",
      "Parameters": {
        "Cluster": "data-pipeline-cluster",
        "TaskDefinition": "load-to-warehouse",
        "LaunchType": "FARGATE"
      },
      "Next": "NotifySuccess"
    },
    "NotifySuccess": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:us-east-1:123456789:pipeline-alerts",
        "Message": "Sales pipeline completed successfully"
      },
      "End": true
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:us-east-1:123456789:pipeline-alerts",
        "Message": "Sales pipeline FAILED"
      },
      "End": true
    }
  }
}

# Schedule using EventBridge (cron)
# Rule: cron(0 2 * * ? *)  # 2 AM daily

Result:
Serverless execution (no infrastructure to manage)
Automatic retries with exponential backoff
Integrates Lambda, Glue, ECS, SNS natively
Visual workflow in AWS Console

✅ Best For

AWS-native stacks
Serverless architectures
Simple to moderate workflows
Event-driven pipelines

❌ Limitations

JSON is verbose, not Pythonic
Limited complex logic
AWS-only (vendor lock-in)
Max 1 year execution time

Azure Data Factory

Cloud ETL and orchestration service for Azure

Azure Data Factory (ADF) is Microsoft's cloud ETL and data integration service. It provides visual pipeline design, 90+ built-in connectors, and deep Azure integration.

Example: ADF Pipeline (Python SDK)

from azure.identity import DefaultAzureCredential
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.mgmt.datafactory.models import *

# Initialize client
credential = DefaultAzureCredential()
adf_client = DataFactoryManagementClient(credential, subscription_id)

# Define pipeline activities
pipeline = {
    "activities": [
        {
            "name": "ExtractSales",
            "type": "Copy",
            "inputs": [{"referenceName": "SourceSQL", "type": "DatasetReference"}],
            "outputs": [{"referenceName": "StagingBlob", "type": "DatasetReference"}],
            "typeProperties": {
                "source": {"type": "SqlSource", "sqlReaderQuery": "SELECT * FROM sales WHERE date = '@{pipeline().parameters.date}'"},
                "sink": {"type": "BlobSink"}
            }
        },
        {
            "name": "TransformSales",
            "type": "DatabricksNotebook",
            "dependsOn": [{"activity": "ExtractSales", "dependencyConditions": ["Succeeded"]}],
            "typeProperties": {
                "notebookPath": "/Notebooks/transform_sales",
                "baseParameters": {
                    "input_path": "@activity('ExtractSales').output.files[0]"
                }
            }
        },
        {
            "name": "LoadToWarehouse",
            "type": "Copy",
            "dependsOn": [{"activity": "TransformSales", "dependencyConditions": ["Succeeded"]}],
            "inputs": [{"referenceName": "TransformedData", "type": "DatasetReference"}],
            "outputs": [{"referenceName": "SynapseWarehouse", "type": "DatasetReference"}]
        },
        {
            "name": "SendNotification",
            "type": "WebActivity",
            "dependsOn": [{"activity": "LoadToWarehouse", "dependencyConditions": ["Succeeded"]}],
            "typeProperties": {
                "url": "https://hooks.slack.com/services/...",
                "method": "POST",
                "body": {"text": "Sales pipeline completed"}
            }
        }
    ]
}

# Create pipeline
adf_client.pipelines.create_or_update(
    resource_group_name="data-rg",
    factory_name="sales-data-factory",
    pipeline_name="sales-etl-pipeline",
    pipeline=pipeline
)

# Create trigger (schedule)
trigger = {
    "type": "ScheduleTrigger",
    "typeProperties": {
        "recurrence": {
            "frequency": "Day",
            "interval": 1,
            "schedule": {"hours": [2], "minutes": [0]}  # 2 AM daily
        }
    }
}

adf_client.triggers.create_or_update(
    resource_group_name="data-rg",
    factory_name="sales-data-factory",
    trigger_name="daily-sales-trigger",
    trigger=trigger
)

Result:
Fully managed, no infrastructure
Visual pipeline designer in Azure Portal
Integrates with Azure Blob, SQL, Databricks, Synapse
Built-in monitoring and alerts

✅ Best For

Azure-native stacks
Teams preferring visual design
Large-scale data movement
Hybrid cloud + on-prem

❌ Limitations

Less flexible than code-first tools
Azure lock-in
Complex pricing model
Limited version control (JSON)

DAG Design Patterns

Common workflow patterns that solve real-world orchestration challenges.

1. Linear Pipeline

Tasks execute in strict sequence. Simple and easy to understand.

Extract → Validate → Transform → Load → Notify

Use case: Simple ETL where each step depends on previous

Most basic pattern, good for straightforward workflows

2. Fan-Out / Fan-In

Execute multiple tasks in parallel, then join results. Speeds up independent operations.

           START
             │
      ┌──────┴──────┬──────┬──────┐
      ▼             ▼      ▼      ▼
  Extract_A   Extract_B  C    Extract_D  (Parallel)
      │             │      │      │
      └──────┬──────┴──────┴──────┘
             ▼
          JOIN & TRANSFORM
             │
             ▼
            LOAD

Use case: Extract from multiple sources simultaneously,
          then combine for processing

Reduces total runtime by parallelizing independent work

Example: Parallel Extraction

from airflow import DAG
from airflow.operators.python import PythonOperator

# Parallel extractions
extract_sales = PythonOperator(task_id='extract_sales', ...)
extract_customers = PythonOperator(task_id='extract_customers', ...)
extract_products = PythonOperator(task_id='extract_products', ...)

# Join and process
join_transform = PythonOperator(
    task_id='join_transform',
    python_callable=lambda: combine_all_data()
)

# Fan-out (parallel)
[extract_sales, extract_customers, extract_products] >> join_transform  # Fan-in

Result: 3 extractions run simultaneously, join waits for all to complete

3. Conditional Branching

Choose execution path based on conditions or data characteristics.

         Extract
            │
            ▼
     Check Data Size
            │
      ┌─────┴─────┐
      │           │
  if LARGE    if SMALL
      │           │
      ▼           ▼
Spark Job    Pandas Job
      │           │
      └─────┬─────┘
            ▼
          Load

Use case: Route data to different processing engines
          based on volume

Optimize resource usage based on runtime conditions

Example: BranchPythonOperator

from airflow.operators.python import BranchPythonOperator

def choose_processing_path(**context):
    """Decide which processing path to take"""
    row_count = context['task_instance'].xcom_pull(task_ids='extract')

    if row_count > 1_000_000:
        return 'process_with_spark'  # Large dataset
    else:
        return 'process_with_pandas'  # Small dataset

branch = BranchPythonOperator(
    task_id='choose_processor',
    python_callable=choose_processing_path
)

spark_task = PythonOperator(task_id='process_with_spark', ...)
pandas_task = PythonOperator(task_id='process_with_pandas', ...)

extract >> branch >> [spark_task, pandas_task]

Result: Automatically routes to Spark for large datasets, Pandas for small

4. Dynamic Task Generation

Generate tasks dynamically based on runtime data (e.g., process one task per table/file).

        List Tables
             │
        ┌────┴────┬────────┬─────────┐
        │         │        │         │
   Process    Process  Process  Process  (Generated at runtime)
   Table_A    Table_B  Table_C  Table_D
        │         │        │         │
        └────┬────┴────────┴─────────┘
             ▼
       Combine Results

Use case: Process each table/file/partition independently

Scale processing to any number of inputs automatically

Example: Dynamic Tasks (Airflow 2.3+)

from airflow.decorators import task

@task
def get_tables():
    """Get list of tables to process"""
    return ['customers', 'orders', 'products', 'reviews']

@task
def process_table(table_name: str):
    """Process a single table"""
    print(f"Processing {table_name}")
    # ETL logic here
    return f"{table_name}_processed"

@task
def combine_results(processed_tables: list):
    """Combine all processed tables"""
    print(f"Combining: {processed_tables}")

# Dynamic pipeline
tables = get_tables()
processed = process_table.expand(table_name=tables)  # Creates N tasks!
combine_results(processed)

Result: Automatically creates 4 parallel tasks (one per table), then combines results

Scheduling, Dependencies & Failure Handling

Scheduling Strategies

Time-Based (Cron)

Run at specific times using cron expressions

0 2 * * * → 2 AM daily
0 */4 * * * → Every 4 hours
0 0 * * 1 → Monday midnight

Event-Based

Trigger when events occur (file arrival, API call, message)

• S3 file upload triggers pipeline
• Kafka message triggers processing
• API webhook starts workflow

Data-Driven

Run when upstream data is ready (sensors)

Wait for file to exist, table to update, or external DAG to complete

Manual

Run on-demand via UI, API, or CLI

Useful for ad-hoc analysis, backfills, or testing

Managing Dependencies

Task Dependencies

# Airflow dependency syntax
task_a >> task_b  # task_b runs after task_a
task_a >> [task_b, task_c]  # task_b and task_c run after task_a (parallel)
[task_a, task_b] >> task_c  # task_c runs after both task_a and task_b

# Alternative syntax
task_b.set_upstream(task_a)
task_c.set_downstream(task_a)

Clear syntax for defining execution order

Cross-DAG Dependencies

Sometimes one pipeline depends on another pipeline completing first.

from airflow.sensors.external_task import ExternalTaskSensor

# Wait for another DAG to complete
wait_for_upstream = ExternalTaskSensor(
    task_id='wait_for_raw_data_pipeline',
    external_dag_id='raw_data_ingestion_dag',
    external_task_id='final_task',
    timeout=3600,  # Wait up to 1 hour
    mode='poke',   # Check every poke_interval
    poke_interval=300  # Check every 5 minutes
)

wait_for_upstream >> start_processing

Result: This DAG waits for raw_data_ingestion_dag to finish before starting

Failure Handling & Retries

Retry Strategies

from datetime import timedelta

task = PythonOperator(
    task_id='flaky_api_call',
    python_callable=call_external_api,

    # Retry configuration
    retries=5,                              # Retry up to 5 times
    retry_delay=timedelta(minutes=5),       # Wait 5 min between retries
    retry_exponential_backoff=True,         # Exponential backoff
    max_retry_delay=timedelta(minutes=60),  # Max 60 min wait

    # Alert configuration
    email_on_retry=False,    # Don't email on retry (too noisy)
    email_on_failure=True,   # Email if all retries fail
    email=['ops@company.com']
)

# Retry timeline:
# Attempt 1: Fails → wait 5 min
# Attempt 2: Fails → wait 10 min (exponential)
# Attempt 3: Fails → wait 20 min
# Attempt 4: Fails → wait 40 min
# Attempt 5: Fails → wait 60 min (capped at max)
# Attempt 6: Fails → email alert, mark as FAILED

Result: Handles transient failures (network issues, rate limits) gracefully

Callbacks & Alerts

def send_slack_alert(context):
    """Send Slack alert on failure"""
    from slack_sdk import WebClient

    task = context.get('task_instance')
    dag = context.get('dag')

    message = f"""
    🚨 *Pipeline Failed*
    DAG: {dag.dag_id}
    Task: {task.task_id}
    Execution Date: {context.get('execution_date')}
    Log: {task.log_url}
    """

    client = WebClient(token=os.environ['SLACK_TOKEN'])
    client.chat_postMessage(channel='#data-alerts', text=message)

# Attach callback to DAG
dag = DAG(
    'critical_pipeline',
    default_args={
        'on_failure_callback': send_slack_alert,
        'on_retry_callback': None,
        'on_success_callback': None
    }
)

Result: Slack notification sent immediately when task fails permanently

Circuit Breaker Pattern

Stop retrying if a dependency is fundamentally broken (e.g., source database down).

from airflow.sensors.base import BaseSensorOperator

class HealthCheckSensor(BaseSensorOperator):
    """Check if upstream system is healthy before proceeding"""

    def poke(self, context):
        try:
            response = requests.get('https://api.source.com/health')
            return response.status_code == 200
        except:
            return False

# Use circuit breaker
health_check = HealthCheckSensor(
    task_id='check_source_health',
    timeout=600,  # Give up after 10 minutes
    poke_interval=60  # Check every minute
)

# Only proceed if healthy
health_check >> extract_from_source >> process >> load

Result: Pipeline waits for source to be healthy, fails fast if unavailable

Why Orchestration is Essential for Production

1. Reliability

Automatic retries, failure handling, and monitoring ensure pipelines don't silently fail. Without orchestration, failed cron jobs go unnoticed until someone checks manually.

2. Visibility

Rich UIs show pipeline status, logs, execution history, and data lineage. Teams can quickly diagnose issues and understand data flow at a glance.

3. Scalability

As pipelines grow from 5 to 500, orchestration manages complexity. Dependency tracking, resource allocation, and parallel execution become impossible to manage manually.

4. Maintainability

Code-based workflows in version control enable collaboration, testing, and rollbacks. Teams can iterate on pipelines without fear of breaking production.

5. Developer Productivity

Pre-built operators, templates, and patterns save time. Developers focus on business logic instead of reinventing scheduling, retries, and logging.

6. Data Quality

Built-in validation, testing, and monitoring catch bad data early. Orchestration platforms like Dagster make data quality a first-class concern.

💡 Bottom Line: Orchestration transforms fragile scripts into production-grade data platforms. The investment in learning and adopting orchestration tools pays dividends in reliability, velocity, and team sanity.

Key Takeaways

Airflow is the industry standard with huge ecosystem
Prefect offers modern developer experience and dynamic workflows
Dagster provides asset-based orchestration with built-in data quality
Cloud services (Step Functions, ADF) integrate seamlessly with cloud stacks

DAG patterns (fan-out, branching, dynamic) solve common workflow challenges
Retries & backoff handle transient failures gracefully
Dependencies ensure tasks execute in correct order
Monitoring & alerting provide visibility and quick incident response

Remember: Choose Airflow for enterprise maturity, Prefect for developer experience, Dagster for data quality focus, or cloud services for simplicity. All are excellent choices; pick based on team skills, scale, and ecosystem fit. Invest in orchestration early, moving from cron to proper orchestration is one of the highest-ROI upgrades in data engineering.

Back to "Data Warehouses"

Big Data FundamentalsLesson 16

Next: "Apache Flink"