Skip to content

Commit edc5fbb

Browse files
committed
fixup! Add user-facing docs for the new Task Loops feature.
The loop docs only told readers what they could not do with a loop task's output from outside the loop, and sent them to publish through a task outside the loop without saying how that task would get the value. Reading the output of a loop task from a task outside the loop now returns every iteration, so describe that: how to refer to the loop task, the sequence the consumer gets, the iteration-then-map-index order for a mapped task, and that a task using a loop's output waits for the whole loop through the loop's gate, with what each trigger rule then sees. The limitations that still hold stay stated plainly: prior-date pulls cannot select a loop iteration, and the wait API still rejects a loop member as the result. Both now point at the recipe that works, a task outside the loop that reads the sequence and returns what it needs.
1 parent cb4d950 commit edc5fbb

1 file changed

Lines changed: 54 additions & 2 deletions

File tree

  • airflow-core/docs/authoring-and-scheduling

‎airflow-core/docs/authoring-and-scheduling/loops.rst‎

Lines changed: 54 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,12 @@ Iterations run one after another; task instances within an iteration can run in
3737
If you are not sure whether you need a loop or mapped tasks, see
3838
:ref:`loops-and-mapped-tasks`.
3939

40+
Before we go any further, it is worth addressing one thing: "How can you have
41+
loops in a Dag, given Dag stands for Directed Acyclic Graph". The answer to that
42+
is by being very careful. Essentially, there is no dependency loop introduced,
43+
the tasks in the Dag still do form a cycle-free graph. The looping nature is
44+
handled at runtime only.
45+
4046
Create a loop
4147
=============
4248

@@ -166,8 +172,8 @@ still running.
166172

167173
Limitations:
168174

169-
* ``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.
170-
* The experimental DagRun wait API also requires an outside-loop result task. It rejects results selected directly from a loop member. See :ref:`dag-result`.
175+
* ``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`.
176+
* 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`.
171177

172178
Connect a loop to other tasks
173179
=============================
@@ -182,6 +188,52 @@ With the default ``all_success`` trigger rule, ``finished`` waits for successful
182188
loop completion. Other trigger rules behave normally; ``always`` does not wait
183189
for the loop.
184190

191+
.. _loops-outside-result:
192+
193+
Use a loop's result outside the loop
194+
====================================
195+
196+
A task outside the loop can read the output of a task inside it. Refer to the
197+
loop task through the loop object and pass its output as an argument:
198+
199+
.. code-block:: python
200+
201+
refinement = refine.loop(max_iterations=10, until=accurate_enough)
202+
203+
204+
@task
205+
def report(results):
206+
print(f"{len(results)} iterations, last error {results[-1]['error']}")
207+
208+
209+
report(refinement["evaluate"].output)
210+
211+
The task receives a sequence (``LazyXComSequence``) with one element for each
212+
iteration the loop ran, in iteration order, so ``results[-1]`` is the last
213+
iteration's result and ``len(results)`` is the number of iterations. A mapped
214+
loop task gives one element per mapped instance for each iteration, ordered by
215+
iteration first and then by map index: all instances of iteration 0, then all
216+
instances of iteration 1, and so on. ``ti.xcom_pull(task_ids=...)`` and
217+
``{{ ti.xcom_pull(task_ids=...) }}`` in a template return the same sequence.
218+
219+
Only the current executions are included. Iterations that a rerun replaced do
220+
not appear.
221+
222+
A task that uses a loop's output waits for the whole loop, not just for the loop task
223+
it reads. Any dependency from a task in a loop to a task outside it, whether from an
224+
argument, ``>>`` or ``set_downstream``, makes the outside task wait for the loop's
225+
gate, which is the task that creates the next iteration. With the default
226+
``all_success`` trigger rule the task then runs after the last iteration and sees
227+
the final sequence. If the loop stops without completing, for example because
228+
the body's terminal task failed or was skipped, the trigger rule applies to the
229+
gate's state. A trigger rule such as ``always`` does not wait for the loop and
230+
may see only the iterations that exist so far.
231+
232+
Inside the loop nothing changes: use ``loop.result`` and ``loop.previous``. A
233+
task outside the loop cannot pull a single value from a loop task, because the
234+
task ran once for each iteration, and cannot use its output to expand a mapped
235+
task.
236+
185237
.. _loops-mapped-tasks:
186238

187239
Mapped tasks inside a loop

0 commit comments

Comments
 (0)