Apache Airflow Guide — DAGs, Operators & Pipeline Orchestration

In this tutorial, you'll learn about Apache Airflow Guide. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Apache Airflow is an open-source workflow Orchestration platform that schedules and monitors Data Pipelines as directed acyclic graphs (DAGs), enabling engineers to automate complex ETL workflows with Python.

What You'll Learn

By the end of this tutorial, you'll understand Airflow's architecture, how to define DAGs and operators, manage task dependencies, schedule pipelines, use sensors for event-driven triggers, and deploy Airflow in production.

Why It Matters

Data Pipelines fail. Sources go down, schemas change, data arrives late. Airflow handles these realities with retries, alerting, backfilling, and dependency management. Without Orchestration, engineers manually restart failed jobs and data teams lose trust in pipeline reliability. DodaTech uses Airflow to orchestrate all Doda Browser telemetry pipelines, processing 500M+ events daily.

Real-World Use

Spotify runs thousands of Airflow DAGs for its recommendation Data Pipelines. Twitter (X) orchestrates its analytics ETL with Airflow. Financial institutions schedule compliance reporting DAGs that must run in sequence with strict SLAs.

Airflow Architecture Overview

flowchart TB
    subgraph "Scheduler"
        S[Scheduler] --> M[(Metadata DB)]
    end
    subgraph "Executor"
        E[Executor] --> W1[Worker 1]
        E --> W2[Worker 2]
        E --> W3[Worker N]
    end
    subgraph "Web"
        W[Web Server] --> M
    end
    subgraph "Storage"
        D[DAG Files] --> S
    end
    W1 --> L[(Logs)]
    W2 --> L
    W3 --> L
    style S fill:#f90,color:#fff
    style W fill:#f90,color:#fff

What Is Apache Airflow?

Airflow models workflows as DAGs (directed acyclic graphs). Each DAG contains tasks (operators) connected by dependencies. The scheduler reads DAG definitions, evaluates task timing, and triggers execution on workers.

A DAG is just a Python file that defines:

• The structure of tasks and their dependencies
• When and how often the DAG should run
• How tasks respond to success, failure, or retries

Core Concepts

DAGs

A DAG is a collection of tasks with directional dependencies. Each task represents a unit of work.

# simple_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

default_args = {
    "owner": "dodatech",
    "depends_on_past": False,
    "email_on_failure": True,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

def extract_data():
    print("Extracting data from source...")
    return {"status": "extracted", "records": 10000}

def transform_data(**context):
    ti = context["ti"]
    extracted = ti.xcom_pull(task_ids="extract")
    print(f"Transforming {extracted['records']} records...")
    return {"status": "transformed", "records": extracted["records"]}

def load_data(**context):
    ti = context["ti"]
    transformed = ti.xcom_pull(task_ids="transform")
    print(f"Loading {transformed['records']} records to warehouse...")
    return {"status": "loaded"}

with DAG(
    dag_id="simple_etl_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    default_args=default_args,
    catchup=False,
    tags=["etl", "dodatech"],
) as dag:
    extract = PythonOperator(task_id="extract", python_callable=extract_data)
    transform = PythonOperator(task_id="transform", python_callable=transform_data)
    load = PythonOperator(task_id="load", python_callable=load_data)

    extract >> transform >> load

Expected output:

[2026-06-23 10:00:00] Running task: extract
[2026-06-23 10:00:01] Extracting data from source...
[2026-06-23 10:00:02] Task extract completed successfully
[2026-06-23 10:00:02] Running task: transform
[2026-06-23 10:00:03] Transforming 10000 records...
[2026-06-23 10:00:04] Task transform completed successfully
[2026-06-23 10:00:04] Running task: load
[2026-06-23 10:00:05] Loading 10000 records to warehouse...
[2026-06-23 10:00:06] Task load completed successfully

Operators

Operators define what each task does. Airflow provides operators for Python, Bash, SQL, sensors, and more.

# operators_demo.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.operators.email import EmailOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator

default_args = {
    "owner": "dodatech",
    "retries": 1,
    "retry_delay": timedelta(minutes=3),
}

with DAG(
    dag_id="operator_comparison",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    default_args=default_args,
    catchup=False,
) as dag:
    run_spark = BashOperator(
        task_id="run_spark_job",
        bash_command="spark-submit --master yarn /jobs/etl.py",
    )

    def process_events(execution_date, **kwargs):
        print(f"Processing events for {execution_date}")
        return {"date": str(execution_date), "events": 150000}

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

    run_query = PostgresOperator(
        task_id="refresh_materialized_view",
        postgres_conn_id="postgres_default",
        sql="REFRESH MATERIALIZED VIEW CONCURRENTLY daily_metrics;",
    )

    run_spark >> process >> run_query

Expected output:

[2026-06-23 10:00:00] Running task: run_spark_job
[2026-06-23 10:00:30] Task run_spark_job completed successfully
[2026-06-23 10:00:31] Running task: process_events
[2026-06-23 10:00:32] Processing events for 2026-06-23
[2026-06-23 10:00:33] Task process_events completed successfully
[2026-06-23 10:00:33] Running task: refresh_materialized_view
[2026-06-23 10:00:35] Task refresh_materialized_view completed successfully

Task Dependencies & Branching

Airflow supports complex dependency patterns including branching, conditional execution, and trigger rules.

# branching_dag.py
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.empty import EmptyOperator
import random

def check_data_quality():
    score = random.uniform(0, 100)
    print(f"Data quality score: {score:.1f}%")
    if score >= 90:
        return ["continue_pipeline"]
    return ["quarantine_data"]

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

with DAG(
    dag_id="quality_branching",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
) as dag:
    start = EmptyOperator(task_id="start")
    quality_check = BranchPythonOperator(
        task_id="quality_check",
        python_callable=check_data_quality,
    )
    continue_pipeline = EmptyOperator(task_id="continue_pipeline")
    quarantine = PythonOperator(
        task_id="quarantine_data",
        python_callable=lambda: print("Moving bad data to quarantine"),
    )
    alert = PythonOperator(
        task_id="alert_team",
        python_callable=alert_team,
        trigger_rule="one_failed",
    )
    end = EmptyOperator(task_id="end", trigger_rule="none_failed_min_one_success")

    start >> quality_check >> [continue_pipeline, quarantine]
    continue_pipeline >> end
    quarantine >> [alert, end]

Expected output:

[2026-06-23 10:00:00] Running task: start
[2026-06-23 10:00:01] Running task: quality_check
[2026-06-23 10:00:02] Data quality score: 73.4%
[2026-06-23 10:00:03] Running task: quarantine_data
[2026-06-23 10:00:04] Moving bad data to quarantine
[2026-06-23 10:00:04] Running task: alert_team
[2026-06-23 10:00:05] Sending alert: data quality below threshold
[2026-06-23 10:00:06] Running task: end

Common Airflow Mistakes

1. Not Setting catchup=False

New Airflow users often forget catchup=False, causing the scheduler to backfill thousands of DAG runs from the start_date. Always set catchup=False unless you explicitly need historical runs.

2. Writing Long-Running Tasks as PythonOperators

If a task takes more than 10 minutes, use a separate worker or KubernetesPodOperator. Long-running PythonOperators block scheduler slots and degrade cluster performance.

3. Ignoring XCom Size Limits

XComs pass data between tasks. By default, they store values in the metadata database, which has a 48KB limit. Use S3, GCS, or a custom backend for large data transfers.

4. No Task Retry Strategy

Tasks fail. Without retries, a transient database timeout takes down your entire pipeline. Set retries=2 and retry_delay=timedelta(minutes=5) on every task.

5. Hard-Coding Connection IDs

Connection strings, passwords, and API keys embedded in DAG files create security risks and break across environments. Always use Airflow's Connections UI and Variable store.

Practice Questions

1. What is a DAG in Apache Airflow and how does it differ from a workflow? A DAG is a directed acyclic graph — tasks with directional dependencies and no cycles. A workflow is a broader concept; a DAG is Airflow's representation of a workflow where tasks must execute in a defined order without circular dependencies.

2. What is the difference between a Sensor and a regular Operator? A Sensor waits for a condition to be met (file appears, API responds, database record exists) before proceeding. It periodically polls and succeeds only when the condition is true. A regular Operator executes work immediately and completes or fails.

3. What are trigger rules and when would you use them? Trigger rules control when a task runs based on upstream task states. Default is all_success. Use all_done to always run regardless of upstream failure, one_failed to run alerts, or none_failed_min_one_success for fan-out patterns.

Frequently Asked Questions

{{< faq question="Can Airflow run tasks in parallel?">}} Yes. Airflow's executor determines parallelism. The LocalExecutor runs tasks in parallel on the same machine using multiprocessing. The CeleryExecutor and KubernetesExecutor distribute tasks across multiple worker nodes for horizontal scaling. Configure parallelism and dag_concurrency in airflow.cfg to control concurrency. {{< /faq >}}

{{< faq question="How do I handle task failures and retries in Airflow?">}} Airflow provides three mechanisms: retries (number of retry attempts), retry_delay (time between retries), and email_on_retry (notification options). You can set these in default_args for all tasks or per-operator. For advanced handling, use on_failure_callback or SLA miss callbacks to trigger alerts via PagerDuty, Slack, or email. {{< /faq >}}

Mini Project: Scheduled Health Check DAG

# health_check_dag.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
import json
import urllib.request

def check_api_health():
    endpoints = {
        "api": "https://api.dodatech.com/health",
        "warehouse": "https://warehouse.dodatech.com/health",
        "auth": "https://auth.dodatech.com/health",
    }
    results = {}
    for name, url in endpoints.items():
        try:
            resp = urllib.request.urlopen(url, timeout=5)
            results[name] = {"status": resp.status, "healthy": resp.status == 200}
        except Exception as e:
            results[name] = {"status": "error", "healthy": False, "error": str(e)}
    total = len(results)
    healthy = sum(1 for r in results.values() if r["healthy"])
    print(f"Health check: {healthy}/{total} services healthy")
    for name, result in results.items():
        icon = "PASS" if result["healthy"] else "FAIL"
        print(f"  [{icon}] {name}: {result['status']}")
    return results

def report_unhealthy(**context):
    ti = context["ti"]
    results = ti.xcom_pull(task_ids="check_health")
    unhealthy = [name for name, r in results.items() if not r["healthy"]]
    if unhealthy:
        print(f"ALERT: Unhealthy services: {', '.join(unhealthy)}")

with DAG(
    dag_id="pipeline_health_check",
    start_date=datetime(2026, 1, 1),
    schedule="*/30 * * * *",
    catchup=False,
    default_args={
        "owner": "dodatech",
        "retries": 1,
        "retry_delay": timedelta(minutes=2),
    },
) as dag:
    check = PythonOperator(task_id="check_health", python_callable=check_api_health)
    report = PythonOperator(
        task_id="report_unhealthy",
        python_callable=report_unhealthy,
        trigger_rule="all_done",
    )
    check >> report

Expected output:

[2026-06-23 10:30:00] Running task: check_health
Health check: 2/3 services healthy
  [PASS] api: 200
  [PASS] warehouse: 200
  [FAIL] auth: 503
[2026-06-23 10:30:01] Running task: report_unhealthy
ALERT: Unhealthy services: auth

Related Concepts

What's Next

You now understand how Airflow orchestrates Data Pipelines with DAGs, operators, and sensors. Next, learn how Apache Spark processes large-scale data transformations, and explore Cloud Computing deployment strategies for Airflow in production.

• Practice daily — Convert a manual Shell Script pipeline into an Airflow DAG
• Build a project — Create a DAG that ingests CSV files, transforms with Python, and loads into PostgreSQL
• Explore related topics — Check out Airflow's TaskFlow API, dynamic DAG generation, and the KubernetesExecutor

Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.