From dd9cac49d1a1a90a42d194483fc3be81dd3ffa35 Mon Sep 17 00:00:00 2001 From: Ash Berlin-Taylor Date: Mon, 5 Oct 2026 10:37:30 +0100 Subject: [PATCH] Add user-facing docs for the new Task Loops feature. Task Loops, introduced in AIP-111, are a whole new way for dag authors to thing Loops chaNge how a Dag author thinks about repeating work, and the change to implement it reaches storage, the scheduler, the Execution API and the UI. Since all these changes are relatively large, it is helpful before we introduct those that reviewers have a mental model of what the behaviour should be, and the features we want to allow are. Rather than explaining this just in PR messages or review threads, lets add this as user facing docs, as most of that will be needed there too. The new Loops doc covers how a loop decides whether to run another pass, how a task reads the previous pass's result, what clearing part of a loop does to the rest, and how loops interact with mapped tasks. At this point/in this PR, nothing exists as code mind you. The short "loop or mapped tasks" page is net-new, and is ther mostly as a jumping off point for "here are the dynamic features you can build your dag with". This oVerview is also why "dynamic mapped tasks" is getting a name change. Mapping is no longer the only way for a Dag's shape to depend on data at runtime, and arguably, mapped tasks aren't really that "dynamic" anymore. "Mapped tasks" is what the UI and the API already call the result ("Clear All Mapped Tasks", MappedOperator), and it reads next to "Loops" as the title of the page the overview links to. "Task mapping" never appears in the UI. However, please note that the rename is "soft" on purpose. The mappIng page keeps its file name and its `dynamic-task-mapping` label, and the old name stays in its opening sentence so existing searches still land. Provider docs link that page by path through intersphinx so moving the file would break their builds until each one is updated. The `dynamic-task-mapping` capability key that the language SDKs report is an identifier rather than prose and is left alone. The snippets are inline here because the example Dags that the guide will include ship with the authoring API, and a file that calls `.loop()` cannot be imported before `.loop()` exists. That change replaces them with included copies that a test executes, so the documented code cannot drift from the behaviour. The placeholder newsfragment is also here because `.loop()` is a new user-visible Dag authoring feature that ships through task-sdk with airflow-core. It will grow and be adjusted by future PRs in this stack. --- .../dag-serialization.rst | 2 +- .../authoring-and-scheduling/deferring.rst | 2 +- .../dynamic-task-mapping.rst | 18 +- .../examples/example_task_loops.py | 114 ++++++ .../docs/authoring-and-scheduling/index.rst | 2 + .../loops-and-mapped-tasks.rst | 61 +++ .../docs/authoring-and-scheduling/loops.rst | 386 ++++++++++++++++++ airflow-core/docs/core-concepts/tasks.rst | 2 +- .../newsfragments/task-loops.feature.rst | 1 + task-sdk/docs/concepts.rst | 2 +- task-sdk/docs/deferred-vs-async-operators.rst | 12 +- task-sdk/docs/dynamic-task-mapping.rst | 8 +- task-sdk/docs/examples.rst | 4 +- 13 files changed, 591 insertions(+), 23 deletions(-) create mode 100644 airflow-core/docs/authoring-and-scheduling/examples/example_task_loops.py create mode 100644 airflow-core/docs/authoring-and-scheduling/loops-and-mapped-tasks.rst create mode 100644 airflow-core/docs/authoring-and-scheduling/loops.rst create mode 100644 airflow-core/newsfragments/task-loops.feature.rst diff --git a/airflow-core/docs/administration-and-deployment/dag-serialization.rst b/airflow-core/docs/administration-and-deployment/dag-serialization.rst index 75bcb2ba379ef..6e3f4f635c6e5 100644 --- a/airflow-core/docs/administration-and-deployment/dag-serialization.rst +++ b/airflow-core/docs/administration-and-deployment/dag-serialization.rst @@ -187,7 +187,7 @@ The system automatically fills in any missing values by walking up the hierarchy MappedOperator Default Handling ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -MappedOperators (dynamic task mapping) also participate in the default value system: +MappedOperators (task mapping) also participate in the default value system: .. code-block:: python diff --git a/airflow-core/docs/authoring-and-scheduling/deferring.rst b/airflow-core/docs/authoring-and-scheduling/deferring.rst index f767cde98675c..9f6c3269506df 100644 --- a/airflow-core/docs/authoring-and-scheduling/deferring.rst +++ b/airflow-core/docs/authoring-and-scheduling/deferring.rst @@ -395,7 +395,7 @@ After the trigger has finished executing, the task may be sent back to the worke start_from_trigger: bool, **kwargs: dict[str, Any], ) -> None: - # This whole method will be skipped during dynamic task mapping. + # This whole method will be skipped during task mapping. super().__init__(*args, **kwargs) self.start_trigger_args.trigger_kwargs = trigger_kwargs diff --git a/airflow-core/docs/authoring-and-scheduling/dynamic-task-mapping.rst b/airflow-core/docs/authoring-and-scheduling/dynamic-task-mapping.rst index 3e2a9e09e80ad..972f744001142 100644 --- a/airflow-core/docs/authoring-and-scheduling/dynamic-task-mapping.rst +++ b/airflow-core/docs/authoring-and-scheduling/dynamic-task-mapping.rst @@ -16,25 +16,29 @@ under the License. .. _dynamic-task-mapping: +.. _mapped-tasks: -==================== -Dynamic Task Mapping -==================== +============ +Mapped tasks +============ -Dynamic Task Mapping allows a way for a workflow to create a number of tasks at runtime based upon current data, rather than the Dag author having to know in advance how many tasks would be needed. +Task mapping, also called dynamic task mapping, allows a way for a workflow to create a number of tasks at runtime based upon current data, rather than the Dag author having to know in advance how many tasks would be needed. This is similar to defining your tasks in a for loop, but instead of having the DAG file fetch the data and do that itself, the scheduler can do this based on the output of an upstream task. -Unlike a Python for-loop executed at DAG parse time, dynamic task mapping defers task creation until runtime, allowing the scheduler to determine the exact number of task instances based on upstream task outputs. +Unlike a Python for-loop executed at DAG parse time, task mapping defers task creation until runtime, allowing the scheduler to determine the exact number of task instances based on upstream task outputs. Right before a mapped task is executed the scheduler will create *n* copies of the task, one for each input. It is also possible to have a task operate on the collected output of a mapped task, commonly known as map and reduce. +.. seealso:: + Not sure whether you need mapped tasks or a loop? See :ref:`loops-and-mapped-tasks`. + Simple mapping ============== In its simplest form you can map over a list defined directly in your Dag file using the ``expand()`` function instead of calling your task directly. -If you want to see a simple usage of Dynamic Task Mapping, you can look below: +If you want to see a simple usage of task mapping, you can look below: .. exampleinclude:: /../src/airflow/example_dags/example_dynamic_task_mapping.py :language: python @@ -388,7 +392,7 @@ For example, this code will *not* work: When code in ``my_task_group`` is executed, ``value`` would still only be a reference, not the real value, so the ``if not value`` branch will not work as you likely want. However, if you pass that reference into a task, it will become resolved when the task is executed, and the three ``my_task`` instances will therefore receive 1, 2, and 3, respectively. -It is, therefore, important to remember that, if you intend to perform any logic on a value passed into a task group function, you must always use a task to run the logic, such as ``@task.branch`` (or ``BranchPythonOperator``) for conditions, and task mapping methods for loops. +It is, therefore, important to remember that, if you intend to perform any logic on a value passed into a task group function, you must always use a task to run the logic, such as ``@task.branch`` (or ``BranchPythonOperator``) for conditions, and ``expand()`` to iterate over values. To repeat a group of tasks, see :ref:`loops-and-mapped-tasks`. .. note:: Task-mapping in a mapped task group is not permitted diff --git a/airflow-core/docs/authoring-and-scheduling/examples/example_task_loops.py b/airflow-core/docs/authoring-and-scheduling/examples/example_task_loops.py new file mode 100644 index 0000000000000..80e41a414680d --- /dev/null +++ b/airflow-core/docs/authoring-and-scheduling/examples/example_task_loops.py @@ -0,0 +1,114 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Repeat task groups with fixed counts, runtime conditions, and mapped tasks.""" + +from __future__ import annotations + +# [START refine_estimate] +from airflow.sdk import dag, task, task_group + + +@dag(schedule=None, catchup=False, tags=["example"]) +def refine_estimate(): + @task_group + def refine(): + @task + def improve(*, loop): + estimate = 1.0 if loop.index == 0 else loop.previous["estimate"] + return (estimate + 2.0 / estimate) / 2.0 + + @task + def evaluate(estimate): + return {"estimate": estimate, "error": abs(estimate * estimate - 2.0)} + + evaluate(improve()) + + def accurate_enough(*, loop): + return loop.result["error"] < 0.000001 + + @task + def finished(): + print("Refinement finished.") + + refinement = refine.loop(max_iterations=10, until=accurate_enough) + refinement >> finished() + + +refine_estimate() +# [END refine_estimate] + + +# [START fixed_loop] +@dag(schedule=None, catchup=False, tags=["example"]) +def fixed_task_loop(): + @task_group + def accumulate(): + @task + def increment(*, loop): + return (0 if loop.index == 0 else loop.previous) + 1 + + increment() + + accumulate.loop(max_iterations=3) + + +fixed_task_loop() +# [END fixed_loop] + + +# [START mapped_loop] +@dag(schedule=None, catchup=False, tags=["example"]) +def mapped_task_loop(): + @task_group + def process_batch(): + @task + def choose_items(*, loop): + return [1, 2] if loop.index == 0 else [value + 1 for value in loop.previous] + + @task + def process(value, *, loop, ti): + print(f"Iteration {loop.index}, mapped position {ti.map_index}") + return value + loop.index + + process.expand(value=choose_items()) + + def batch_ready(*, loop): + return min(loop.result) >= 2 + + process_batch.loop(max_iterations=3, until=batch_ready) + + +mapped_task_loop() +# [END mapped_loop] + + +# [START partial_override_loop] +@dag(schedule=None, catchup=False, tags=["example"]) +def partial_override_loop(): + @task_group + def accumulate(increment_by): + @task + def increment(increment_by, *, loop): + return (0 if loop.index == 0 else loop.previous) + increment_by + + increment(increment_by) + + accumulate.override(group_id="accumulate_more").partial(increment_by=5).loop(max_iterations=3) + + +partial_override_loop() +# [END partial_override_loop] diff --git a/airflow-core/docs/authoring-and-scheduling/index.rst b/airflow-core/docs/authoring-and-scheduling/index.rst index d91e57eafe79b..67a7ab19f0d1e 100644 --- a/airflow-core/docs/authoring-and-scheduling/index.rst +++ b/airflow-core/docs/authoring-and-scheduling/index.rst @@ -31,7 +31,9 @@ It's recommended that you first review the pages in :doc:`core concepts diff --git a/airflow-core/docs/authoring-and-scheduling/loops-and-mapped-tasks.rst b/airflow-core/docs/authoring-and-scheduling/loops-and-mapped-tasks.rst new file mode 100644 index 0000000000000..7b8621313cab2 --- /dev/null +++ b/airflow-core/docs/authoring-and-scheduling/loops-and-mapped-tasks.rst @@ -0,0 +1,61 @@ + .. Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + .. http://www.apache.org/licenses/LICENSE-2.0 + + .. Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + +.. _loops-and-mapped-tasks: + +====================== +Loops and mapped tasks +====================== + +A Dag does not have to be static. How much work it does can depend on data that only exists at +runtime: how many files arrived, what the previous step found, or whether an answer is good enough +yet. Airflow has two features for this, as well as mechanisms like ``retries`` and dynamic Dag generation that fit adjacent use cases. + +Choose a mechanism +================== + +.. list-table:: + :header-rows: 1 + :widths: 30 30 40 + + * - You want to + - Use + - How it works + * - Run the same task once for each item in a collection + - :ref:`Mapped tasks ` + - ``task.expand(...)`` creates one task instance per item once the collection is known. The + instances can run concurrently. + * - Repeat a group of tasks, each iteration building on the result of the previous one, until a + condition is met + - :ref:`Loops ` + - ``task_group.loop(...)`` creates the next iteration of task instances only if the last iteration says + another is needed. + * - Give a failed task another try + - Retries, set with ``retries`` on the task. See :doc:`/core-concepts/tasks`. + - A retry reruns the same task instance. It does not create new task instances, and it does not advance a + loop. + * - Create tasks from something known when the Dag file is parsed + - A Python ``for`` loop in the Dag file. See :doc:`/howto/dynamic-dag-generation`. + - The Dag has the same shape in every run. + +Use them together +================= + +An iteration of a loop can contain mapped tasks, so each iteration can fan out over a different collection of +items. See :ref:`Mapped tasks inside a loop `. + +Mapping a whole loop, nesting loops, and placing a loop inside a mapped task group are not supported. diff --git a/airflow-core/docs/authoring-and-scheduling/loops.rst b/airflow-core/docs/authoring-and-scheduling/loops.rst new file mode 100644 index 0000000000000..8cd0c76f06d01 --- /dev/null +++ b/airflow-core/docs/authoring-and-scheduling/loops.rst @@ -0,0 +1,386 @@ + .. Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + .. http://www.apache.org/licenses/LICENSE-2.0 + + .. Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + +.. _loops: + +===== +Loops +===== + +A loop repeats a group of tasks a number of times decided at runtime, +using the results of each iteration. Each iteration can use the previous +iteration's result. Use a loop for work such as refining an answer, following +a hierarchy, or processing successive batches. + +A conditional loop works like a ``do … while`` loop: run the body, then test +whether to run it again. Airflow's ``until`` condition expresses when to stop, +so it corresponds to ``do { body } while not until(result)``. The body runs at +least once. ``max_iterations`` sets an upper limit; the runtime condition +determines how many iterations are needed within that limit. + +Iterations run one after another; task instances within an iteration can run in parallel. + +If you are not sure whether you need a loop or mapped tasks, see +:ref:`loops-and-mapped-tasks`. + +Before we go any further, it is worth addressing one thing: "How can you have +loops in a Dag, given Dag stands for Directed Acyclic Graph". The answer to that +is by being very careful. Essentially, there is no dependency loop introduced, +the tasks in the Dag still do form a cycle-free graph. The looping nature is +handled at runtime only. + +Create a loop +============= + +Define the body with ``@task_group`` and call ``.loop()`` on the decorated +function. Supply a positive integer ``max_iterations`` to limit the number of +iterations. An optional ``until`` callable decides when to stop early. + +Airflow evaluates ``until`` in a gate task downstream of the body's terminal task. +It must return a ``bool``: + +* ``True`` means stop: the condition has been met. +* ``False`` means continue, provided another iteration is allowed. + +Any other return value, including the ``None`` that a forgotten ``return`` produces, will result in +the gate task failing, and the loop not continuing. + +Airflow creates the next iteration when the gate says to continue. +It does not create all ``max_iterations`` up front: +if the loop stops after four iterations, you see those four, without a tail +of skipped iterations. A fixed-count loop runs exactly ``max_iterations`` +iterations. + +This example improves an estimate of the square root of two until the error +is small enough: + +.. exampleinclude:: /authoring-and-scheduling/examples/example_task_loops.py + :start-after: [START refine_estimate] + :end-before: [END refine_estimate] + +``evaluate`` returns the result for the iteration. The gate reads it through +``loop.result``. If another iteration runs, ``improve`` reads that same result +through ``loop.previous``. The gate appears as a task in the loop, named after +the condition function: ``refine.accurate_enough`` in this example. + +Tasks inside a loop body receive ``loop`` as a context parameter, so a parameter with that name in a +loop body must be keyword-only and cannot have a default other than ``None``. + +In a templated field, a Jinja ``{% for %}`` block defines its own ``loop`` that hides this one for +the length of the block. Read the Airflow value before the block and use that inside it, for example +``{% set iteration = loop.index %}``. + +When ``until`` has no usable name, the gate is named ``__loop_gate`` instead. +That covers a lambda, a ``functools.partial`` and a callable object. A gate name +that matches a task in the loop body gets a ``__1`` suffix, as with any other +task ID. + +Stopping because ``until`` returned ``True`` means the loop converged. +Reaching ``max_iterations`` while the condition remains ``False`` means the +loop did not converge: the gate task in the final iteration is marked as +failed. Meeting the condition on the last allowed iteration marks the gate task as success. + +For a fixed-count loop, omit ``until``. The definition ``accumulate.loop(max_iterations=3)`` runs +three iterations, carrying results between them. Reaching the cap completes +a fixed-count loop successfully. Its gate is named ``__loop_gate`` within the group. + +.. exampleinclude:: /authoring-and-scheduling/examples/example_task_loops.py + :start-after: [START fixed_loop] + :end-before: [END fixed_loop] + +For a task-group function with arguments, supply them with ``.partial()`` before +calling ``.loop()``. Use ``.override()`` to configure the group, for example to +give another loop a different ``group_id``: + +.. exampleinclude:: /authoring-and-scheduling/examples/example_task_loops.py + :start-after: [START partial_override_loop] + :end-before: [END partial_override_loop] + +Read the loop context +===================== + +Declare ``loop`` as a keyword-only parameter on a task function. Airflow +supplies it at execution time; leave it out when calling the task in the Dag +definition. + +.. list-table:: Loop context + :header-rows: 1 + :widths: 25 75 + + * - Attribute + - Meaning + * - ``loop.index`` + - The current iteration number, starting at zero. + * - ``loop.max_iterations`` + - The configured limit on the number of iterations. + * - ``loop.previous`` + - The previous iteration's result, or ``None`` in iteration 0. + * - ``loop.result`` + - The terminal body task's result (the XCom pushed under the key ``return_value``) for the current iteration, used by the gate. + +Use ``loop.index`` for the loop iteration and ``ti.map_index`` for a mapped +task instance's position. Each task instance can have several tries, each with its own ``ti.try_number``. + +Pass data between iterations +============================ + +Between iterations, use ``loop.previous``. Check ``loop.index == 0`` to handle +the first iteration. Do not test ``loop.previous is None``: it is also ``None`` +when the previous terminal task returned nothing, and a previous result of +``0``, ``False``, or an empty collection can be valid data. + +The loop body must have exactly one terminal task definition before the gate. +Its return value (the XCom pushed under the ``return_value`` key) becomes ``loop.result`` for the gate and ``loop.previous`` +for the next iteration. A mapped terminal task supplies its collection of +results as a sequence (``LazyXComSequence``), the same as other downstream consumers of mapped tasks. + +If the body has several branches, finish with a task that combines their +outputs. For example, if two branches end in ``refine_left`` and +``refine_right``, add this inside the task group: + +.. code-block:: python + + @task + def combine(left, right): + return {"left": left, "right": right} + + + combine(refine_left(), refine_right()) + +``combine`` is now the single terminal task. The gate can read each result +through ``loop.result["left"]`` and ``loop.result["right"]``; tasks in the +next iteration use the corresponding keys in ``loop.previous``. + +Setup and teardown tasks may be used in the body, but teardowns do not count as +terminal tasks and the gate does not wait for them. A failing teardown does not stop the +loop, and the next iteration can start while the previous iteration's teardowns are +still running. + +Limitations: + +* ``include_prior_dates=True`` cannot select a loop iteration from another Dag run. Push the result to XCom through a task outside the loop if later Dag runs need to retrieve it with an ordinary XCom pull; see :ref:`loops-outside-result`. +* The experimental DagRun wait API also requires an outside-loop result task. Marking a loop member as the Dag result raises ``ValueError`` while the Dag is parsed, and the wait API rejects results selected directly from a loop member. Mark a task outside the loop as the result instead, for example one that returns the last element of a loop task's results. See :ref:`dag-result`. + +Connect a loop to other tasks +============================= + +Use the object returned by ``.loop()`` in dependencies: + +.. code-block:: python + + refinement >> finished() + +With the default ``all_success`` trigger rule, ``finished`` waits for successful +loop completion. Other trigger rules behave normally; ``always`` does not wait +for the loop. + +.. _loops-outside-result: + +Use a loop's result outside the loop +==================================== + +A task outside the loop can read the output of a task inside it. Refer to the +loop task through the loop object and pass its output as an argument: + +.. code-block:: python + + refinement = refine.loop(max_iterations=10, until=accurate_enough) + + + @task + def report(results): + print(f"{len(results)} iterations, last error {results[-1]['error']}") + + + report(refinement["evaluate"].output) + +The task receives a sequence (``LazyXComSequence``) with one element for each +iteration the loop ran, in iteration order, so ``results[-1]`` is the last +iteration's result and ``len(results)`` is the number of iterations. A mapped +loop task gives one element per mapped instance for each iteration, ordered by +iteration first and then by map index: all instances of iteration 0, then all +instances of iteration 1, and so on. ``ti.xcom_pull(task_ids=...)`` and +``{{ ti.xcom_pull(task_ids=...) }}`` in a template return the same sequence. + +Only the current executions are included. Iterations that a rerun replaced do +not appear. + +A task that uses a loop's output waits for the whole loop, not just for the loop task +it reads. Any dependency from a task in a loop to a task outside it, whether from an +argument, ``>>`` or ``set_downstream``, makes the outside task wait for the loop's +gate, which is the task that creates the next iteration. With the default +``all_success`` trigger rule the task then runs after the last iteration and sees +the final sequence. If the loop stops without completing, for example because +the body's terminal task failed or was skipped, the trigger rule applies to the +gate's state. A trigger rule such as ``always`` does not wait for the loop and +may see only the iterations that exist so far. + +Inside the loop nothing changes: use ``loop.result`` and ``loop.previous``. A +task outside the loop cannot pull a single value from a loop task, because the +task ran once for each iteration, and cannot use its output to expand a mapped +task. + +.. _loops-mapped-tasks: + +Mapped tasks inside a loop +========================== + +A loop iteration can contain mapped tasks. Use mapping to process several +items within the iteration. The next iteration can operate on a different +collection of items. + +A mapped task can be the body's single terminal task: + +.. code-block:: text + + select items → process each item → gate + +The gate reads the collection of mapped results through ``loop.result``. Add +a combining task before the gate if you want to reduce those results to a +single value or structure. A zero-length expansion skips the mapped task and, +under the gate's ``all_success`` rule, skips the gate; no next iteration is +created. + +In this example, ``choose_items`` returns the values to map over: ``[1, 2]`` in the +first iteration, then each previous result plus one. + +.. exampleinclude:: /authoring-and-scheduling/examples/example_task_loops.py + :start-after: [START mapped_loop] + :end-before: [END mapped_loop] + +The loop iteration and mapped position are separate "coordinates". In this +example, each iteration has mapped instances 0 and 1, so mapped instance 1 in +iteration 0 is distinct from mapped instance 1 in iteration 1. +Mapped instances appear within their loop iteration, so you can inspect their +states, tries, and logs separately. + +Mapping a whole loop, nesting loops, and placing a loop inside a mapped task +group are not supported. + +Failures, skips, and retries +============================ + +A retry stays in the same iteration and does not consume another iteration. +Tasks in the body follow normal trigger rules. For example, a final combining +task with ``all_done`` can handle a branch failure and return a result. If that +task succeeds, the gate evaluates its result normally. If the terminal task +fails, the gate does not run under the default ``all_success`` rule, and +no next iteration is created. The gate has no settings of its own: it takes +``trigger_rule`` and ``retries`` from the Dag's ``default_args`` like any other +task. With ``trigger_rule="all_done"`` in ``default_args``, the gate runs after +a failed terminal task, and a fixed-count loop continues because it only +counts iterations. + +An exception in ``until`` fails the gate task and appears in its logs; the +gate is retried according to ``retries``. An ``until`` that returns anything +other than a ``bool``, and a loop that reaches its iteration limit without +meeting ``until``, fail the gate without retrying, because a retry would see the +same result. In the second case the final gate fails; downstream tasks with the +default ``all_success`` trigger rule will not run, as described under +non-convergence above. + +Skipping the body's terminal task also skips the gate under the default +``all_success`` trigger rule, so no next iteration is created. Downstream +tasks with the default ``all_success`` rule are skipped too; give a downstream +task ``trigger_rule="none_failed"`` if it should still run when the loop ends +without running: + +.. code-block:: python + + @task(trigger_rule="none_failed") + def finished(): ... + + + refinement >> finished() + +If some branches may be skipped but the loop should continue, give the final +combining task a trigger rule that permits those skips and have it return the +iteration's result. + +Manually marking a gate successful completes it without evaluating ``until`` +or creating another iteration. This is an explicit override of normal gate +execution. Downstream tasks then run as they would after any successful gate: they cannot tell a +gate that was marked successful by hand from one that stopped because ``until`` returned ``True``. +Any later iterations retained after a selective clear remain unchanged. + +Iterations and execution history +================================ + +In the Grid, select the loop's task group in a Dag run to open its Task Instances +tab. The **Iteration** filter narrows the table to one iteration, or shows +**All iterations**, and the **Iteration** column shows which iteration each task +instance belongs to. The gate's state and logs explain why the loop continued, +stopped, or failed. Each task's tries and logs remain accessible within its +iteration. + +Iterations cleared by a rerun are kept as history but are not shown in the UI +in 3.4.0. + +Clear tasks inside a loop +========================== + +Use these controls to select how far a clear extends through the loop: + +* **Clear downstream tasks** includes downstream tasks. It starts selected unless you have + changed your saved clear options. Clearing a task this way can also clear + the gate in the same iteration. +* **Clear later loop iterations**, selected by default, clears later + iterations when the selection includes a gate task, whether selected + directly or through **Clear downstream tasks**. The gate runs again and decides + whether the loop should continue from that point. + +Clearing later iterations does not require the loop to run the same number of +iterations again. The gate can stop earlier or continue further, within the +configured limit. + +Rerun part of an iteration +--------------------------- + +Suppose each iteration contains this sequence: + +.. code-block:: text + + prepare → process → consume → gate + +The loop has already run iterations 0 through 4. You clear ``process`` in +iteration 2 with **Clear downstream tasks** and **Clear later loop iterations** both selected: + +* ``prepare`` in iteration 2 remains completed. +* ``process``, ``consume``, and the gate in iteration 2 run again. +* Later iterations 3 and 4 are cleared. Whether replacement iterations run + depends on the new gate decisions. +* Iterations 0 and 1 remain untouched. + +If the gate now stops at iteration 2, replacement iterations 3 and 4 are not +created. + +Keep later iterations +--------------------- + +Deselect **Clear later loop iterations** to keep the later iterations. +Rerunning an earlier gate then does not change the loop progression that +already followed it: a new stop result does not remove later iterations, and +a new continue result does not create a duplicate next iteration. + +The choice is the same wherever you clear. The REST clear endpoint takes it as +``include_later_loop_iterations``, which defaults to ``true``, matching the +checked box in the UI. ``airflow tasks clear`` has no option for it: it always +clears later iterations, except with ``--only-failed`` or ``--only-running``, +which leave them in place. To keep later iterations, clear from the UI or the +REST endpoint. + +Tasks you leave uncleared are not recomputed using the results of the rerun. diff --git a/airflow-core/docs/core-concepts/tasks.rst b/airflow-core/docs/core-concepts/tasks.rst index e56fed21e371a..00f095aa66e4c 100644 --- a/airflow-core/docs/core-concepts/tasks.rst +++ b/airflow-core/docs/core-concepts/tasks.rst @@ -275,7 +275,7 @@ Define a policy once and share it across DAGs via ``default_args`` or a shared m Mapped tasks ~~~~~~~~~~~~ -Policies work with dynamic task mapping. The policy applies per mapped task +Policies work with task mapping. The policy applies per mapped task instance -- if instance 2 of 10 hits FAIL, the other 9 continue independently: .. code-block:: python diff --git a/airflow-core/newsfragments/task-loops.feature.rst b/airflow-core/newsfragments/task-loops.feature.rst new file mode 100644 index 0000000000000..385d11d75a3d7 --- /dev/null +++ b/airflow-core/newsfragments/task-loops.feature.rst @@ -0,0 +1 @@ +Repeat decorated task groups with runtime loop conditions and selective clearing using ``.loop()``. diff --git a/task-sdk/docs/concepts.rst b/task-sdk/docs/concepts.rst index a314830f36ee8..0e499956c6ee6 100644 --- a/task-sdk/docs/concepts.rst +++ b/task-sdk/docs/concepts.rst @@ -45,7 +45,7 @@ Understanding the task lifecycle helps Dag authors write more effective tasks an - ``task_reschedule_count``: Number of times this task has been rescheduled - ``variables``: List of Airflow variables accessible to the task instance - ``connections``: List of Airflow connections accessible to the task instance - - ``upstream_map_indexes``: Mapping of upstream task IDs to their map indexes for dynamic task mapping scenarios + - ``upstream_map_indexes``: Mapping of upstream task IDs to their map indexes for task mapping scenarios - ``next_method``: Method name to call when resuming from a deferred state (set when task resumes from a trigger) - ``next_kwargs``: Arguments to pass to the ``next_method`` (can be encrypted for sensitive data) - ``xcom_keys_to_clear``: List of XCom keys that need to be cleared and purged by the worker diff --git a/task-sdk/docs/deferred-vs-async-operators.rst b/task-sdk/docs/deferred-vs-async-operators.rst index 4212e36246109..9429aff552d3e 100644 --- a/task-sdk/docs/deferred-vs-async-operators.rst +++ b/task-sdk/docs/deferred-vs-async-operators.rst @@ -265,17 +265,17 @@ it is equally important to understand scenarios where one may **not** be appropr If your workflow is better suited for deferring but no operator exists yet, consider implementing a custom deferred operator rather than defaulting to async. -Comparison with Dynamic Task Mapping ------------------------------------- +Comparison with task mapping +---------------------------- -Async operators and Dynamic Task Mapping solve different problems and have different trade-offs. +Async operators and task mapping solve different problems and have different trade-offs. .. list-table:: :header-rows: 1 * - Aspect - Async ``@task`` - - Dynamic Task Mapping (deferrable) + - Task mapping (deferrable) * - Worker slots - 1 worker slot (shared event loop) - N worker slots (one per mapped task instance) @@ -292,5 +292,5 @@ Async operators and Dynamic Task Mapping solve different problems and have diffe - Minimal - Scheduler must manage N task instances -For more details about Dynamic Task Mapping, see the -:ref:`dynamic task mapping ` page. +For more details about task mapping, see the +:ref:`task mapping ` page. diff --git a/task-sdk/docs/dynamic-task-mapping.rst b/task-sdk/docs/dynamic-task-mapping.rst index 84481bffcac81..60f1b31bb09a3 100644 --- a/task-sdk/docs/dynamic-task-mapping.rst +++ b/task-sdk/docs/dynamic-task-mapping.rst @@ -17,10 +17,10 @@ .. _sdk-dynamic-task-mapping: -Dynamic Task Mapping with Task SDK -================================== +Mapped tasks with Task SDK +========================== -Dynamic Task Mapping allows tasks defined with the Task SDK to generate +Task mapping, also called dynamic task mapping, allows tasks defined with the Task SDK to generate a variable number of task instances at runtime based on upstream data. This is enabled via the ``expand()`` method on tasks, providing a way to parallelize execution without knowing the number of tasks ahead of time. @@ -96,4 +96,4 @@ For advanced patterns—such as repeated mapping, cross-product mapping, named mappings (via ``map_index_template``), and handling large datasets—see the Airflow Core documentation: -`Dynamic Task Mapping in the Airflow Core docs `_. +`Mapped tasks in the Airflow Core docs `_. diff --git a/task-sdk/docs/examples.rst b/task-sdk/docs/examples.rst index a3115d79a90bf..7cf50bca78929 100644 --- a/task-sdk/docs/examples.rst +++ b/task-sdk/docs/examples.rst @@ -63,13 +63,13 @@ Tasks and Operators Example: Defining tasks and using operators -Use the :func:`airflow.sdk.task` decorator to wrap Python callables as tasks and leverage dynamic task mapping with the ``.expand()`` method. Tasks communicate via :class:`airflow.sdk.XComArg`. For traditional operators and sensors, import classes like :class:`airflow.sdk.BaseOperator` or :class:`airflow.sdk.Sensor`. +Use the :func:`airflow.sdk.task` decorator to wrap Python callables as tasks and leverage task mapping with the ``.expand()`` method. Tasks communicate via :class:`airflow.sdk.XComArg`. For traditional operators and sensors, import classes like :class:`airflow.sdk.BaseOperator` or :class:`airflow.sdk.Sensor`. .. exampleinclude:: ../../airflow-core/src/airflow/example_dags/example_dynamic_task_mapping.py :language: python :start-after: [START example_dynamic_task_mapping] :end-before: [END example_dynamic_task_mapping] - :caption: Dynamic task mapping with ``expand()`` + :caption: Task mapping with ``expand()`` .. exampleinclude:: ../../airflow-core/src/airflow/example_dags/example_xcomargs.py :language: python