A mapped task instance is one execution-specific occurrence of an Airflow task, created for a single element of dynamic mapping input. If a ratings API returns three reviews, the mapped task produces three independently processable instances. section 12.1.8
Use partial() for parameters shared by every instance and expand() for the values being mapped. The number of instances can differ between DAG runs, enabling parallel processing while preserving the upstream-to-downstream relationship. section 12.1.8
Listing 12.32 Using Dynamic Task Mapping to generate tasks
def _fetch_ratings():
data = requests.get("http://movie-reviews:8081/reviews/latest")
return [[x] for x in data.json()] #1
def _print_rating(rating):
print(f"New rating for Movie:
{rating["movie"]}. Rating: {rating["rating"]}")
with DAG(
dag_id="05_dynamic_task_mapping",
start_date=pendulum.today("UTC").add(days=-5),
schedule=CronTriggerTimetable("0 16 * * *", timezone="UTC")
):
fetch_ratings = PythonOperator(
task_id="fetch_ratings",
python_callable=_fetch_ratings
)
print_rating = PythonOperator.partial( #2
task_id="print_rating",
python_callable=_print_rating
).expand(op_args=fetch_ratings.output) #3
fetch_ratings >> print_rating
This example maps print_rating over fetch_ratings.output, creating one rating-processing instance per returned review. section 12.1.8
Operationally, mapped instances are visible as a group and individually:
- The Airflow grid and graph views mark mapped tasks with [] and expose their individual instances. section 12.1.8
- TaskGroups can also be mapped, showing [] at both task and group levels. section 12.1.8
