Tasks: Behavior & Customization

13Retry policy#

What it does: Configures how many times a failed task automatically retries, and how long it waits between attempts.

Parameters:

How it works:

• retries=2 means 3 total attempts — the first try plus 2 retries.
• With retry_exponential_backoff=True: delay = min(retry_delay_seconds * 2^(try_number-1), max_retry_delay_seconds), plus a random ±10% jitter to avoid many tasks retrying at exactly the same moment. If retry_delay_seconds is 0 or unset, it's floored to 1 second before the exponential math runs.
• Without max_retry_delay_seconds, exponential backoff has no ceiling — a task with many retries can end up waiting a long time between later attempts.
• Set retry defaults once for the whole DAG via default_args=, and override per task only where needed — a task-level value always wins over the DAG default.
• A task that's voluntarily skipped (feature 34) does not consume a retry attempt — only genuine failures do.

Example:

# DAG-wide defaults via default_args...
with DAG(
    dag_id="resilient_pipeline",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    default_args={
        "retries": 3,
        "retry_delay_seconds": 30,
        "retry_exponential_backoff": True,
        "max_retry_delay_seconds": 600,
    },
) as dag:

    # ...inherits retries=3, exponential backoff, 30s base, 600s cap
    default_behavior_task = PythonOperator(task_id="a", python_callable=lambda: None)

    # ...overrides just the retry count for this one task, keeps the rest
    flaky_task = PythonOperator(
        task_id="b",
        python_callable=lambda: None,
        retries=5,
    )

14Trigger rules#

What it does: Decides when a task becomes eligible to run, based on the states of its upstream tasks — instead of the default "every upstream must succeed."

Parameters:

How it works:

• A task with no upstream dependencies is always immediately ready, whatever trigger_rule you set — the rule only matters once there's at least one upstream edge.
• always runs a task unconditionally — including if every upstream failed or was skipped. Handy for cleanup/notification steps.
• none_failed_min_one_success is stricter than none_failed: it additionally requires at least one upstream to have actually succeeded, not just "no failures." A branch where every upstream was skipped satisfies none_failed but not none_failed_min_one_success.
• all_done_setup_success is all_done plus requiring any upstream is_setup tasks to have succeeded (feature 20).
• Type your trigger_rule string carefully — an unrecognized value quietly falls back to all_success semantics rather than raising an error, which can look like a logic bug in your DAG rather than a typo.

Example:

with DAG(dag_id="branch_convergence", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    extract = PythonOperator(task_id="extract", python_callable=lambda: None)
    branch_a = PythonOperator(task_id="branch_a", python_callable=lambda: None)
    branch_b = PythonOperator(task_id="branch_b", python_callable=lambda: None)

    # Runs once EITHER branch finishes successfully, without waiting for both
    join = PythonOperator(
        task_id="join",
        python_callable=lambda: None,
        trigger_rule="one_success",
    )

    # Always runs a cleanup step, even if upstream tasks failed/were skipped
    cleanup = PythonOperator(
        task_id="cleanup",
        python_callable=lambda: None,
        trigger_rule="always",
    )

    extract >> [branch_a, branch_b] >> join
    [branch_a, branch_b] >> cleanup

15Cross-run dependencies#

What it does: Gates a task on the state of the same task (or its downstream) from the DAG's previous run — useful for keeping sequential runs from racing ahead of unfinished prior work.

Parameters:

How it works:

• Both flags compare against the immediately preceding run of the same DAG, not "the last successful run." If there is no previous run to compare against (e.g. this is the DAG's first run), the check is effectively a no-op.
• wait_for_downstream=True on a task with no downstream tasks has nothing to wait on — it behaves as if it were False.
• Both can be set once via default_args and overridden per task, using the same inheritance pattern as retries (feature 13).
• Combining depends_on_past=True with max_active_runs=1 effectively forces fully serial execution of that task's lineage across runs.

Example:

with DAG(
    dag_id="sequential_state_machine",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
) as dag:

    # Won't start until this same task succeeded in the PREVIOUS run
    load_incremental = PythonOperator(
        task_id="load_incremental",
        python_callable=lambda: None,
        depends_on_past=True,
    )

    # Won't start until every downstream task from the PREVIOUS run's
    # "publish" task finished, preventing overlapping publishes
    publish = PythonOperator(
        task_id="publish",
        python_callable=lambda: None,
        wait_for_downstream=True,
    )

    load_incremental >> publish

16Execution timeout#

What it does: Kills a single task attempt if it runs past a set number of seconds.

Parameters:

How it works:

• If unset, a task isn't unbounded — it falls back to the deployment's globally configured default task timeout.
• On expiry, the task is hard-killed (not a cooperative cancellation) — anything it was mid-write on may be left partially done if it wasn't written atomically.
• A timeout counts as a normal failure for retry purposes — set retries if you want a timed-out attempt to get another chance.
• This is a task-level timeout, independent of the DAG-level dagrun_timeout_seconds (feature 6) — a task can time out well before the whole run's timeout is reached.

Example:

slow_but_bounded = PythonOperator(
    task_id="slow_but_bounded",
    python_callable=lambda: None,
    execution_timeout=300,   # kill this task if it runs past 5 minutes
    retries=1,               # give it one more shot if it times out
)

17Task-level SLA#

What it does: Flags an SLA miss if a task hasn't reached a terminal state within a set number of seconds of the DAG run's logical date. Detection only — never kills or fails the task.

Parameters:

How it works:

• The clock starts at the DAG run's logical date, not the task's own start time — a task scheduled later in a chain effectively has less of its own SLA budget left, since earlier tasks already consumed part of the same window.
• You get at most one SLA-miss record per task instance, no matter how many times it's re-evaluated on later scheduler ticks.
• There's no separate per-task SLA-miss callback — a miss reuses the DAG's on_sla_miss_callback (feature 7), falling back to the DAG's on_failure_callback if no SLA callback is set.

Example:

critical_step = PythonOperator(
    task_id="critical_step",
    python_callable=lambda: None,
    sla=120,   # flag an SLA miss if not done within 2 minutes of the run's logical date
)

18Concurrency & pools#

What it does: pool assigns a task to a named, shared concurrency budget; task_concurrency caps how many instances of that specific task can run at once across all active runs of its DAG.

Parameters:

How it works:

• The "default" pool is seeded with 128 slots system-wide — every task that doesn't opt into a different pool shares that same budget with everything else in the deployment.
• Assigning a pool name that doesn't exist yet doesn't create it — ask an admin to create the pool first if you need a real, smaller concurrency ceiling.
• Pool slots are shared across all DAGs referencing the same pool name — a very active DAG can crowd out a lower-volume DAG sharing the same pool.
• task_concurrency is independent of, and applies simultaneously with, max_active_tasks (feature 5) — a task needs a free slot under both (plus its pool) to be dispatched.
• Hitting a limit never fails or skips a task — it just waits, still scheduled, for a slot to open up.

Example:

heavy_task = PythonOperator(
    task_id="heavy_transform",
    python_callable=lambda: None,
    pool="etl_heavy",        # ask an admin to create this pool for a real cap
    task_concurrency=2,      # never more than 2 concurrent instances of THIS task_id
)

19Prioritization#

What it does: Ranks tasks against each other when there's contention for dispatch — under normal, uncongested conditions it has no visible effect.

Parameters:

How it works:

• Dispatch order under load is highest priority_weight first. Under system load-shedding, low or negative weights get dropped before high ones.
• weight_rule="upstream": effective weight = this task's own weight plus the sum of every ancestor's own weight, computed recursively — a task deep in a long chain automatically inherits a compounding boost.
• weight_rule="downstream": mirror image — effective weight = own weight plus the sum of every descendant's weight — a task gating a large fan-out automatically gets a bigger boost.
• This computation is a real dependency-graph walk performed on every planning pass for any DAG using a non-"absolute" rule — cheap for normal-sized DAGs, but genuine work, not a cached one-time value.

Example:

with DAG(dag_id="priority_demo", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    # Absolute (default): weight is exactly 50, unaffected by graph position
    urgent = PythonOperator(task_id="urgent", python_callable=lambda: None, priority_weight=50)

    # Upstream: effective weight = 10 + sum of every ancestor's own priority_weight
    gate = PythonOperator(
        task_id="gate", python_callable=lambda: None,
        priority_weight=10, weight_rule="upstream",
    )

    urgent >> gate

20Lifecycle role: setup & teardown#

What it does: Marks a task as preparing resources (is_setup) or cleaning them up (is_teardown). Teardown tasks get special treatment — they're triggered once every "real work" task in the run is done, regardless of how those tasks turned out.

Parameters:

How it works:

• Teardown tasks are never scheduled through the normal dependency-graph pass, even once their upstreams are satisfied — they're only triggered later, in bulk, once every non-teardown task has reached a terminal state. In practice this means a teardown task runs regardless of whether upstream tasks succeeded, failed, or were skipped.
• Because of that, it's common (though not required) to declare trigger_rule="all_done" on a teardown task, just to document the intent clearly in code.
• is_setup=True doesn't change scheduling on its own — a setup task follows the normal graph and its own trigger_rule like any other task. Its only special effect is enabling all_done_setup_success for tasks downstream of it.
• Setting both is_setup and is_teardown on the same task isn't meaningful — the teardown behavior takes over.

Example:

with DAG(dag_id="managed_cluster_job", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    spin_up = PythonOperator(
        task_id="spin_up_cluster",
        python_callable=lambda: None,
        is_setup=True,
    )

    process = PythonOperator(task_id="process_data", python_callable=lambda: None)

    tear_down = PythonOperator(
        task_id="tear_down_cluster",
        python_callable=lambda: None,
        is_teardown=True,
        trigger_rule="all_done",   # documents intent
    )

    spin_up >> process >> tear_down

21Per-task callbacks#

What it does: Fires an alert per individual task instance's own outcome — success, failure, retry, or skip — independent of the DAG-level callbacks in feature 9.

Parameters:

You can configure a task-level callback two ways:

1. Directly on the operator: on_success_callback / on_failure_callback / on_retry_callback / on_skipped_callback, each a plain Python function receiving a context dict — use this for custom logic.
2. Declaratively, via params={"_callbacks": {event: config}}, where event is one of on_success, on_failure, on_retry, on_skipped, and config is a dict describing an email/Slack/HTTP webhook/PagerDuty alert — use this when you want a built-in channel without writing any Python (see Part 7 for the exact shape of each channel's config).

How it works:

• Use the direct on_X_callback=my_function kwargs for custom Python logic (log something, call an internal API, write a task note, etc.).
• Use the params={"_callbacks": {...}} dict form for a built-in email/Slack/HTTP webhook/PagerDuty alert with no custom code.
• If you don't set a given event's callback, nothing is added for that event.
• These stack with DAG-wide default_args the same way retries do (feature 13) — a task-level callback wins; otherwise the DAG's default (if any) is inherited.
• on_skipped only fires for the task's own terminal skip — it does not fire for a task that was skipped by branch evaluation (feature 33) or by trigger-rule skip-propagation (feature 14) — those are considered "never actually executed," so there's no outcome to alert on.
• on_execute_callback (fires when a task starts) is accepted syntactically but has no observable effect today — it isn't part of this event set.
• If you need a DAG-wide "final outcome, whatever it was" alert, add a dedicated notifier task with trigger_rule="all_done" (feature 46) rather than relying on a callback.

Example:

def log_retry(context):
    print(f"retrying: {context['task_id']}")

flaky_call = PythonOperator(
    task_id="flaky_call",
    python_callable=lambda: None,
    retries=2,
    on_retry_callback=log_retry,
    params={
        "_callbacks": {
            "on_failure": {"type": "slack", "connection_id": "slack_alerts_webhook"},
        }
    },
)

22Environment control#

What it does: Controls what a Python/Bash task's subprocess actually sees: extra environment variables, which interpreter runs it, and (for PythonVirtualenvOperator) which managed virtualenv it runs under.

Parameters:

How it works:

• By design, tasks never inherit the orchestrator's own environment — you get a small fixed allowlist (PATH, HOME, TZ, LANG, LC_ALL, DAGS_REPO_PATH, PYTHONPATH) by default. Orchestrator secrets are never part of that list, so there's no way to leak them through env=, even by naming them explicitly.
• append_env=False gives a "pristine" subprocess: only the bare essentials (PATH, HOME, DAGS_REPO_PATH, PYTHONPATH) plus whatever you put in env= — note that even in pristine mode, TZ/LANG/LC_ALL are dropped unless you set them yourself.
• python=/venv= re-import your DAG module fresh inside the target interpreter (no pickling of the callable) — the target interpreter must be able to import dag_parser and any third-party packages your callable needs.
• A named venv= environment must already be built by an admin before any task can use it — a DAG referencing one that was never built will fail when the task actually runs, not at ingestion time.

Example:

# env + append_env — overlay mode (default): allowlist + your vars
BashOperator(
    task_id="with_extra_env",
    bash_command="echo $STAGE $PATH",
    env={"STAGE": "staging"},
    append_env=True,   # default; PATH/HOME/TZ/... still present alongside STAGE
)

# ExternalPythonOperator — run under a specific pre-provisioned interpreter
ExternalPythonOperator(
    task_id="run_under_custom_interpreter",
    python_callable=lambda: None,
    python="/opt/piflow/venv-extra/bin/python3",
)

# PythonVirtualenvOperator — named pre-built managed venv
PythonVirtualenvOperator(
    task_id="run_in_managed_venv",
    python_callable=lambda: None,
    venv="pandas_env",   # must already be built by an admin
)

# PythonVirtualenvOperator —  requirements (auto-derived env name)
PythonVirtualenvOperator(
    task_id="run_with_requirements",
    python_callable=lambda: None,
    requirements=["pandas==2.2.0", "requests==2.31.0"],
)