How can we specify an upstream task for all copies of a task created by

.map

?

Hey Luke - you can use the following pattern:

from prefect.utilities.tasks import unmapped

...

result = map_task.map(..., upstream_tasks=[unmapped(upstream_task)])