DAGs, Operators & Dependencies

Defining a DAG

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

with DAG(
    dag_id='daily_etl',
    schedule='@daily',        # cron expression or preset; None = manual-trigger only
    start_date=datetime(2026, 1, 1),
    catchup=False,             # don't backfill every missed interval since start_date
    default_args={'retries': 3, 'retry_delay': timedelta(minutes=5)},
) as dag:

    extract = PythonOperator(task_id='extract', python_callable=extract_data)
    transform = PythonOperator(task_id='transform', python_callable=transform_data)
    load = BashOperator(task_id='load', bash_command='python load.py')

    extract >> transform >> load   # upstream >> downstream dependency

XCom — Passing Small Data Between Tasks

def extract_data(**context):
    file_path = upload_to_s3(fetch_raw_data())
    context['ti'].xcom_push(key='s3_path', value=file_path)  # small reference, not the data itself

def transform_data(**context):
    file_path = context['ti'].xcom_pull(task_ids='extract', key='s3_path')
    data = download_from_s3(file_path)
    # ...

# XCom lives in the metadata database — pass a REFERENCE (a path, an ID),
# never large datasets themselves, which would degrade metadata DB performance

Sensors & Manual Triggers

# CLI
airflow dags trigger daily_etl
airflow dags backfill daily_etl -s 2026-01-01 -e 2026-01-07

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id='wait_for_upload',
    filepath='/data/incoming/daily.csv',
    poke_interval=60,
)
wait_for_file >> extract