Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
2 changes: 1 addition & 1 deletion airflow-core/docs/authoring-and-scheduling/deferring.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Comment thread
ashb marked this conversation as resolved.

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
Expand Down Expand Up @@ -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

Expand Down
Comment thread
ashb marked this conversation as resolved.
Original file line number Diff line number Diff line change
@@ -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]
2 changes: 2 additions & 0 deletions airflow-core/docs/authoring-and-scheduling/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,9 @@ It's recommended that you first review the pages in :doc:`core concepts </core-c
deferring
serializers
connections
loops-and-mapped-tasks
dynamic-task-mapping
loops
assets
dag-result
Non-Python Task SDKs <language-sdks/index>
Expand Down
Original file line number Diff line number Diff line change
@@ -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 <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 <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 <loops-mapped-tasks>`.

Mapping a whole loop, nesting loops, and placing a loop inside a mapped task group are not supported.
Loading