MWAA pre-installs only Airflow core, the Amazon provider package, and their transitive dependencies. Do not assume any other Python package (pandas, pyarrow, numpy, scikit-learn, etc.) is available at task runtime. If DAG code needs a package not in the base image:

Add it to requirements.txt (installed at environment startup). Pin every package to an exact version and install against the Apache Airflow constraints file that matches the environment’s Airflow and Python version (--constraint "https://raw.githubusercontent.com/apache/airflow/constraints-<AIRFLOW_VERSION>/constraints-<PY_VERSION>.txt", with the exact URL fetched at runtime). Unpinned or unconstrained requirements are the most common cause of a failed environment update — a transitive dependency resolves to a version incompatible with the pinned Airflow.
Or delegate the work to a managed service (Glue for Spark/pandas, SageMaker for ML, Lambda for lightweight transforms) that has its own runtime.

A ModuleNotFoundError at task execution (not at parse time) means the import is inside a task callable and the package is not installed. The fix is requirements.txt or delegation — not inlining the install in the DAG.

Timeout Chain

sensor.timeout <= dagrun_timeout
service_timeout <= dagrun_timeout
execution_timeout = safety net per non-deferred window (15min typical)
dagrun_timeout >= sum(service timeouts) + retry buffer

For deferrable operators, execution_timeout resets on each deferred-to-active cycle. It need not exceed the service timeout.

Retry Categories

Category	retries	retry_delay	Notes
Sensor	0	--	Has own timeout
API call	3	5min	Short, transient failures
Compute job (expensive)	1	5min	Costly reruns; minimize
Compute job (cheap idempotent)	2	5min	Safe to retry
DQ gate	2	5min	AirflowFailException non-retryable
Teardown	2	5min	trigger_rule=all_done

Always: retry_exponential_backoff=True, max_retry_delay=timedelta(minutes=30).

Failure Callback Template

def notify_failure(context):
    import json
    import boto3
    from datetime import datetime, timezone

    task_instance = context["task_instance"]
    dag_id = context["dag"].dag_id
    task_id = task_instance.task_id
    start_date = task_instance.start_date
    end_date = task_instance.end_date or datetime.now(timezone.utc)
    exception = context.get("exception", "")

    message = {
        "dag_id": dag_id,
        "task_id": task_id,
        "start_date": str(start_date),
        "end_date": str(end_date),
        # Publish only the exception TYPE (carries no secret) + the log_url;
        # never the message/traceback/args, which can carry secrets or PII
        # (CWE-532). Authorized viewers get full detail via log_url.
        "error_type": type(exception).__name__ if exception else "",
        "log_url": task_instance.log_url,
    }

    sns = boto3.client("sns", region_name=REGION)
    sns.publish(
        TopicArn=SNS_TOPIC_ARN,
        Subject=f"Airflow task failed: {dag_id}.{task_id}",
        Message=json.dumps(message, indent=2),
    )

Security: enable server-side encryption (SSE) on the SNS topic, restrict Publish to the execution role, and confirm SNS_TOPIC_ARN is an expected in-account recipient — the callback sends operational metadata off the environment.

Waiting for AWS Resource State Changes

Exhaust Airflow-native constructs before writing custom wait code. An operator or sensor that owns the wait is almost always preferable to a hand-written boto3 poll — it frees the worker slot and encodes the polling and error handling for you. Work down this ladder and stop at the first rung that applies:

Deferrable operator. If a deferrable operator exists for the action, use it with deferrable=True (e.g., EmrServerlessStartJobOperator(deferrable=True)). It hands the wait to the triggerer, freeing the worker slot for the whole wait and resuming on the completion event.
Submit + deferrable sensor. If no single operator both submits and waits, submit with a non-blocking operator (e.g., wait_for_completion=False) and wait with a deferrable sensor (deferrable=True). Deferrable support is per-sensor and per-provider-version — verify it for the specific sensor. Some AWS sensors are NOT deferrable (e.g. AthenaSensor, EmrServerlessJobSensor); for those, use the deferrable operator form instead (AthenaOperator(deferrable=True), EmrServerlessStartJobOperator(deferrable=True)). If no deferrable path exists, a sensor in mode="reschedule" still frees the worker slot between pokes (no triggerer needed, but it only resumes on a fixed interval).
Bounded in-task poll loop. Only when no operator or sensor fits and you must wait inside a @task with boto3. Do not assume client.get_waiter("<state>") exists — not every service/operation ships a waiter, and a missing one raises ValueError: Waiter does not exist at runtime. Verify waiter availability in the boto3 docs for your runtime’s botocore version before using get_waiter; otherwise use a bounded loop:

for _ in range(max_attempts):
    response = client.get_<resource>(<id>=resource_id)
    state = response["<resource>"]["state"]
    if state == target_state:
        break
    if state in terminal_failure_states:
        raise RuntimeError(f"Resource reached {state}")
    time.sleep(delay_seconds)
else:
    raise RuntimeError(f"Resource did not reach {target_state}")

Deferrable operators and sensors require a running triggerer; MWAA runs a managed triggerer on all MWAA-supported Airflow versions, so the deferrable rungs work out of the box on both Airflow 2 and 3. The mode="reschedule" fallback needs no triggerer. See A6 rule 5 in authoring-provisioned-dag.md for the submit + @task sensor split.

start_date

Use a fixed, timezone-aware datetime from stdlib. Never use days_ago() (deprecated in 2.8, removed in 3.x) or naive datetimes (rejected in 3.x). Always pair a fixed past start_date with catchup=False: on Airflow 2.x catchup defaults to True, so a scheduled DAG with a past start_date backfills every missed interval in one burst the moment it is unpaused. AF3 defaults catchup=False, but set it explicitly so the DAG behaves the same on both.

from datetime import datetime, timezone

with DAG(
    dag_id="example",
    start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
    schedule="<mm> <h> * * *",  # off-peak cron DERIVED from dag_id (A6 rule 15): mm 1-59, h 1-6 UTC — recompute per DAG, never a fixed literal or @preset
    catchup=False,  # REQUIRED with a fixed past start_date — avoids a backfill storm
):
    ...

Do not use pendulum.datetime(...) — it adds a top-level dependency on pendulum’s API which changed between v2 and v3. Stdlib timezone.utc works on all MWAA-supported Python versions (3.10+) and all Airflow versions (2.x and 3.x).

Idempotency

Job names with "{{ ds_nodash }}" suffixes for date uniqueness.

Cross-DAG Triggering

Event-driven coupling. The object type and its in-task event key differ by Airflow major, so gate the example by the target Airflow version (per A3 “verify APIs exist in the Airflow version”):

# Airflow 2.4-2.x (Datasets):
from airflow.datasets import Dataset
DATASET = Dataset("s3://bucket/prefix/")
# Producer: outlets=[DATASET] on final task
# Consumer: schedule=[DATASET]
# In-task events: context["triggering_dataset_events"]

# Airflow 3.x (Assets replace Datasets):
from airflow.sdk import Asset
ASSET = Asset("s3://bucket/prefix/")
# Producer: outlets=[ASSET] on final task
# Consumer: schedule=[ASSET]
# In-task events: context["triggering_asset_events"]

Cross-environment: MwaaTriggerDagRunOperator if support available in Airflow version in use.

references/dag-patterns.md

Raw file

aws/agent-toolkit-for-aws