https://prefect.io logo
Join SlackCommunities
Powered by
is there an equivalent to task.map but runs the ma...
# ask-community
j
is there an equivalent to task.map but runs the mapped parameter elements in serial instead? basically identical to map but forced to run in order. I looked at the loop signal as an option, but I don't think that will lay out the schematic in the UI as nicely as map does
and I would prefer to build the iterable param in a task, not outside of the task environment
a
Yes, if you use mapping with the default sequential executor, it will run your mapped tasks sequentially. Here is an example:
Copy code
from prefect import Flow, task


@task(log_stdout=True)
def add_ten(x):
    print(x)
    return x


with Flow("mapping-sequentially") as flow:
    mapped_result = add_ten.map([1, 2, 3, 4, 5])

if __name__ == "__main__":
    flow.run()
the output is 1, 2, 3, 4, 5 one after the other:
Copy code
[2022-01-12 18:38:26+0100] INFO - prefect.FlowRunner | Beginning Flow run for 'mapping-sequentially'
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten': Starting task run...
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten': Finished task run for task with final state: 'Mapped'
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[0]': Starting task run...
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | 1
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[0]': Finished task run for task with final state: 'Success'
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[1]': Starting task run...
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | 2
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[1]': Finished task run for task with final state: 'Success'
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[2]': Starting task run...
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | 3
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[2]': Finished task run for task with final state: 'Success'
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[3]': Starting task run...
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | 4
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[3]': Finished task run for task with final state: 'Success'
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[4]': Starting task run...
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | 5
[2022-01-12 18:38:26+0100] INFO - prefect.TaskRunner | Task 'add_ten[4]': Finished task run for task with final state: 'Success'
[2022-01-12 18:38:26+0100] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
k
Looping would be the closest thing here because looping would run the input sequentially. I think having to build the iterable inside the task also falls in line with mapping
j
I see. I think the executor in production is not sequential, so loop signal might have to be the way to go. not sure what that will look like in the UI once I'm ready to deploy, but hopefully they are shown as atomic tasks per loop iteration
k
They are not because that technically wouldnt be a DAG.
j
so the task with the loop invocation will show up as a single task in the schematic after everything finishes, instead of a task per iteration? if that's the case, it sounds like the loop indication will only appear in logs, in which case I can probably just use a for loop internally and not deal with signals altogether
k
I think that’s right. I can test in a bit to be sure
🙏 1
Yeah this is right. For Orion (Prefect 2.0) though, this type of dynamicism is supported because there is no DAG requirement
j
great! thanks for checking. what's the best way to follow along on Orion updates?
k
It would be the release notes here
2 Views
Open in Slack
PreviousNext
Powered by