Airflow best practices: a practical guide for developers

By day 47 in your Spark and data engineering journey, you've likely built a few Airflow DAGs—and discovered that not all of them scale gracefully. Airflow best practices aren't just about avoiding crashes; they're about building reliable DAGs that your team can maintain, debug, and extend without constant firefighting. This guide distils production lessons into concrete patterns you can apply today.

Keep DAG files lightweight

Your DAG file is parsed by the Airflow scheduler every dag_dir_list_interval seconds (default 300). If parsing takes too long, you risk scheduler lag and delayed task execution. Many teams make the mistake of loading entire datasets, running queries, or initialising heavy libraries at the module level.

Consider a typical anti-pattern:

from airflow import DAG
from airflow.operators.python import PythonOperator
import pandas as pd
from datetime import datetime

# WRONG: This executes every time the DAG file is parsed
df = pd.read_csv('s3://my-bucket/large-dataset.csv')
config = run_expensive_query()  # Database call at parse time

def process_data():
    return df.groupby('id').sum()

dag = DAG('heavy_dag', start_date=datetime(2024, 1, 1))
task = PythonOperator(
    task_id='process',
    python_callable=process_data,
    dag=dag
)

Every time the scheduler parses this DAG—which could be dozens of times per minute across your deployment—it loads a CSV and runs a query. This bloats memory, creates database load, and slows the scheduler.

The fix is straightforward: defer all I/O and computation to task execution time, not DAG definition time.

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

def load_and_process():
    import pandas as pd
    # Now this happens only when the task runs
    df = pd.read_csv('s3://my-bucket/large-dataset.csv')
    config = run_expensive_query()
    return df.groupby('id').sum()

dag = DAG('light_dag', start_date=datetime(2024, 1, 1))
task = PythonOperator(
    task_id='process',
    python_callable=load_and_process,
    dag=dag
)

Your DAG file should only contain operators, dependencies, and lightweight configuration. Move logic, imports, and I/O into task callables. This keeps your scheduler responsive and your DAG parsing time under control—typically under 100ms per DAG.

Idempotent tasks

Airflow will retry failed tasks. If a task fails midway and retries, you need to ensure running it again produces the same result without duplication or inconsistency. This is idempotency, and it's essential for reliable DAGs.

Consider a task that inserts records into a database:

from airflow.operators.python import PythonOperator
from datetime import datetime

def insert_events():
    conn = get_db_connection()
    cursor = conn.cursor()
    events = fetch_events_from_api()
    for event in events:
        cursor.execute(
            "INSERT INTO events (id, timestamp, data) VALUES (%s, %s, %s)",
            (event['id'], event['timestamp'], event['data'])
        )
    conn.commit()

task = PythonOperator(
    task_id='insert_events',
    python_callable=insert_events,
    retries=3,
    retry_delay=timedelta(minutes=5)
)

If the network fails after inserting 50 out of 100 events, the retry will attempt to insert all 100 again, creating duplicates. Not idempotent.

Make it idempotent by using INSERT ... ON CONFLICT or MERGE (depending on your database), or by deleting and re-inserting the data for the execution date:

def insert_events_idempotent(execution_date):
    conn = get_db_connection()
    cursor = conn.cursor()
    
    # Delete any existing data for this execution date
    cursor.execute(
        "DELETE FROM events WHERE DATE(timestamp) = %s",
        (execution_date.date(),)
    )
    
    # Now insert fresh
    events = fetch_events_from_api()
    for event in events:
        cursor.execute(
            "INSERT INTO events (id, timestamp, data) VALUES (%s, %s, %s)",
            (event['id'], event['timestamp'], event['data'])
        )
    conn.commit()

task = PythonOperator(
    task_id='insert_events',
    python_callable=insert_events_idempotent,
    op_args=[datetime(2024, 1, 15)],  # or use {{ ds }} template
    retries=3,
    retry_delay=timedelta(minutes=5)
)

Alternatively, use a database-native approach:

def insert_events_upsert():
    conn = get_db_connection()
    cursor = conn.cursor()
    events = fetch_events_from_api()
    for event in events:
        cursor.execute("""
            INSERT INTO events (id, timestamp, data) 
            VALUES (%s, %s, %s)
            ON CONFLICT (id) DO UPDATE 
            SET timestamp = EXCLUDED.timestamp, data = EXCLUDED.data
        """, (event['id'], event['timestamp'], event['data']))
    conn.commit()

Idempotent tasks make retries safe. They're a prerequisite for a production system.

Use templated dates

Airflow provides templated variables like {{ ds }} (date string in YYYY-MM-DD), {{ ds_nodash }}, {{ execution_date }}, and more. These are rendered at runtime based on the DAG's execution context. Using templates keeps your DAG dynamic and avoids hardcoding dates.

Here's why templating matters. Without it, you might write:

from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

dag = DAG(
    'daily_report',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
)

# WRONG: Hardcoded, won't work for backfill
task = BashOperator(
    task_id='generate_report',
    bash_command='python /scripts/report.py --date 2024-01-15',
    dag=dag
)

If you want to backfill data for January, you'd have to manually change the date. This doesn't scale.

With templating:

from airflow.operators.bash import BashOperator
from datetime import datetime

dag = DAG(
    'daily_report',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily'
)

task = BashOperator(
    task_id='generate_report',
    bash_command='python /scripts/report.py --date {{ ds }}',
    dag=dag
)

Now when this DAG runs on 2024-01-15, {{ ds }} becomes 2024-01-15. On 2024-01-16, it becomes 2024-01-16. Backfilling automatically uses the correct dates.

You can also pass templates to Python callables:

def process_daily_data(ds):
    """ds is passed as a template variable"""
    print(f"Processing data for {ds}")
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM raw_events WHERE DATE(created_at) = %s",
        (ds,)
    )
    return cursor.fetchall()

task = PythonOperator(
    task_id='process',
    python_callable=process_daily_data,
    op_kwargs={'ds': '{{ ds }}'},  # Passed as a template
    dag=dag
)

Templating ensures your DAG works correctly across all execution dates, whether running once or backfilling six months of data.

Avoid passing large data through XCom

XCom (cross-communication) is Airflow's mechanism for passing data between tasks. By default, XCom data is serialised (usually as JSON) and stored in the Airflow metadata database. It's convenient but has significant limitations.

Consider this pattern:

def extract_large_dataset():
    df = pd.read_csv('s3://bucket/huge-file.csv')  # 500MB
    return df.to_dict('records')  # Returns as Python objects

def transform_data(ti):
    records = ti.xcom_pull(task_ids='extract')  # Fetch from XCom
    # Process records
    return len(records)

task1 = PythonOperator(
    task_id='extract',
    python_callable=extract_large_dataset,
    dag=dag
)

task2 = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    dag=dag
)

task1 >> task2

When extract_large_dataset returns a huge dictionary, Airflow serialises it to JSON, writes it to the database, and then deserialises it in the next task. This is slow, bloats your metadata database, and can fail silently if the serialised size exceeds database limits (often 64KB or 1GB depending on backend).

Instead, pass only metadata through XCom and store the data externally:

def extract_large_dataset():
    df = pd.read_csv('s3://bucket/huge-file.csv')
    # Write to S3, not XCom
    output_path = 's3://bucket/processed/extract_{{ ds }}.parquet'
    df.to_parquet(output_path)
    return {'path': output_path, 'rows': len(df)}

def transform_data(ti):
    metadata = ti.xcom_pull(task_ids='extract')
    df = pd.read_parquet(metadata['path'])  # Load from S3
    # Process
    return len(df)

task1 = PythonOperator(
    task_id='extract',
    python_callable=extract_large_dataset,
    dag=dag
)

task2 = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    dag=dag
)

task1 >> task2

Now XCom carries only a small metadata dictionary, and the actual data lives in S3. This is faster, scales to terabytes, and keeps your metadata database lean. Use XCom for configuration, paths, and counts—not for datasets.

Common mistakes

• Parsing-time computation: Running database queries, API calls, or loading large files at the module level instead of inside task functions.
• Non-idempotent inserts: Not handling retries properly, leading to duplicate data or failed backfills.
• Hardcoded dates: Writing dates directly into SQL queries or bash commands instead of using {{ ds }} templates, making backfills painful.
• XCom for big data: Returning entire DataFrames through XCom, which bloats the metadata database and triggers serialisation errors.
• Missing task dependencies: Forgetting to set explicit dependencies (with >> or <<), leading to tasks running out of order or in parallel unexpectedly.
• Ignoring task timeouts: Not setting execution_timeout, so a hung task blocks downstream tasks indefinitely.

Key takeaways

• Keep your DAG files lightweight—defer all I/O and computation to task execution time. Your scheduler will thank you.
• Design tasks to be idempotent so retries don't cause data inconsistencies or duplicates. Use upserts or delete-and-reinsert strategies.
• Use Airflow's templated variables like {{ ds }} to make your DAGs dynamic and backfill-friendly.
• Pass only metadata through XCom; store large datasets in S3, HDFS, or your data warehouse. Keep the metadata database lean.
• Set explicit task dependencies, execution timeouts, and retry policies to build resilient, debuggable DAGs.
• Test your DAGs locally with the SequentialExecutor and validate logic before deploying to production.