Writing DAGs

TaskFlow API (Modern Way)

The TaskFlow API (Airflow 2.0+) uses Python decorators. It's cleaner than classic operators and handles XCom automatically.

from airflow.decorators import dag, task
from datetime import datetime

@dag(
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["example"]
)
def my_etl():

    @task
    def extract():
        """Pull data from API"""
        return {"users": 100, "revenue": 5000}

    @task
    def transform(raw_data: dict):
        """Clean and enrich"""
        raw_data["avg_revenue"] = raw_data["revenue"] / raw_data["users"]
        return raw_data

    @task
    def load(clean_data: dict):
        """Write to warehouse"""
        print(f"Loading: {clean_data}")

    # Dependencies are inferred from function calls!
    raw = extract()
    clean = transform(raw)
    load(clean)

my_etl()  # Instantiate the DAG

Branching — Conditional Paths

Use @task.branch to choose which downstream path to follow at runtime.

@task.branch
def check_data_quality(row_count: int):
    if row_count > 0:
        return "process_data"    # task_id to run
    else:
        return "send_alert"      # skip process_data, run alert

@task
def process_data():
    print("Processing...")

@task
def send_alert():
    print("No data! Alerting team.")

count = extract()
check_data_quality(count) >> [process_data(), send_alert()]

Sensors — Wait for Events

Sensors are special operators that wait until a condition is true before allowing downstream tasks to run.

from airflow.sensors.filesystem import FileSensor
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

# Wait for a file to appear
wait_for_file = FileSensor(
    task_id="wait_for_file",
    filepath="/data/incoming/daily_export.csv",
    poke_interval=60,       # Check every 60 seconds
    timeout=3600,            # Give up after 1 hour
    mode="poke"              # or "reschedule" to free the worker slot
)

# Wait for S3 object
wait_for_s3 = S3KeySensor(
    task_id="wait_for_s3",
    bucket_key="s3://my-bucket/data/{{ ds }}/export.parquet",
    mode="reschedule"        # Frees the slot between pokes
)

Trigger Rules

By default, a task runs only when ALL upstream tasks succeed. Trigger rules let you change this.

from airflow.utils.trigger_rule import TriggerRule

cleanup = PythonOperator(
    task_id="cleanup",
    python_callable=cleanup_temp_files,
    trigger_rule=TriggerRule.ALL_DONE  # Always runs
)

Dynamic DAGs

Since DAGs are Python, you can generate tasks dynamically with loops.

tables = ["users", "orders", "products", "events"]

@dag(schedule="@daily", start_date=datetime(2024, 1, 1), catchup=False)
def ingest_all_tables():

    @task
    def extract_table(table_name: str):
        print(f"Extracting {table_name}")
        return f"s3://bucket/{table_name}.parquet"

    @task
    def load_table(table_name: str, path: str):
        print(f"Loading {table_name} from {path}")

    for table in tables:
        path = extract_table(table)
        load_table(table, path)

ingest_all_tables()

@task
def get_tables():
    return ["users", "orders", "products"]

@task
def process(table: str):
    print(f"Processing {table}")

tables = get_tables()
process.expand(table=tables)  # Creates N parallel tasks at runtime

Best Practices