Idempotency, Scaling & Configuration

Task Idempotency

# BAD — unconditional INSERT duplicates rows if this task retries or re-runs
def load_bad(**context):
    db.execute("INSERT INTO daily_totals (date, total) VALUES (%s, %s)", (date, total))

# GOOD — safe to retry/re-run/backfill any number of times, same end result
def load_good(**context):
    db.execute("""
        INSERT INTO daily_totals (date, total) VALUES (%s, %s)
        ON CONFLICT (date) DO UPDATE SET total = EXCLUDED.total
    """, (date, total))
# Retries and manual re-runs are NORMAL, expected parts of how Airflow
# operates — design every task assuming it might run more than once.

Avoiding Expensive DAG-Parse-Time Code

# BAD — the Scheduler re-parses every DAG file repeatedly to detect
# changes; this API call re-runs on EVERY parse, not just once
config = requests.get('https://api.example.com/config').json()

with DAG(...) as dag:
    task = PythonOperator(task_id='use_config', python_callable=lambda: process(config))

# GOOD — expensive work happens only when the TASK actually runs
def use_config():
    config = requests.get('https://api.example.com/config').json()
    process(config)

with DAG(...) as dag:
    task = PythonOperator(task_id='use_config', python_callable=use_config)

Connections & Variables

from airflow.hooks.base import BaseHook
from airflow.models import Variable

conn = BaseHook.get_connection('my_postgres')  # credentials stored once, not hardcoded
api_base_url = Variable.get('api_base_url')     # editable without a code deploy

Scaling: Celery vs. Kubernetes Executor

The Celery Executor distributes tasks to a pool of persistently-running workers via a message broker — efficient for stable, predictable load. The Kubernetes Executor launches a fresh pod per task instance — stronger per-task isolation and elastic scaling, at the cost of per-task pod-startup overhead. Choosing between them depends on the workload's actual resource and isolation needs.