Hi, I have a question about

apply_map

. Details following in the thread 🙂

I think it works. The example in docs uses it here . Maybe don’t call the

run()

?

This is the error message if I remove

run()

`Task 'iterate_files[0]': Exception encountered during task execution! Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 876, in get_task_run_state value = prefect.utilities.executors.run_task_with_timeout( File "/usr/local/lib/python3.9/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout return task.run(*args, **kwargs) # type: ignore File "/Users/marius/Documents/Projekte/Pexon/FMC/project-panorama/prefect-experimentation/5008-flow/flow.py", line 44, in iterate_files File "/usr/local/lib/python3.9/site-packages/prefect/core/task.py", line 633, in call new.bind( File "/usr/local/lib/python3.9/site-packages/prefect/core/task.py", line 693, in bind raise ValueError( ValueError: Could not infer an active Flow context while creating edge to <Task: parse_recfile>. This often means you called a task outside a

with Flow(...)

block. If you're trying to run this task outside of a Flow context, you need to call `parse_recfile.run(...)``

That is because you have the task decorator still above the

iterate_files

If I remove the task decorator, it seems like it calls the function while registering the flow, not while running. I get this error message:

Traceback (most recent call last):

File "/Users/marius/Documents/Projekte/Pexon/FMC/project-panorama/prefect-experimentation/5008-flow/flow.py", line 113, in <module>

infos = apply_map(iterate_files, file=files, zip_filename=unmapped(zip_filename))

File "/Users/marius/miniforge3/envs/prefect-fmc/lib/python3.9/site-packages/prefect/utilities/tasks.py", line 131, in apply_map

res = func(*args2, **kwargs2)

File "/Users/marius/Documents/Projekte/Pexon/FMC/project-panorama/prefect-experimentation/5008-flow/flow.py", line 41, in iterate_files

is_recfile = file[0].endswith("...")

AttributeError: 'GetItem' object has no attribute 'endswith'

That is because the content of the function has to be tasks but this line:

is_recfile = file[0].endswith("...")

is not a task and it executes eagerly. A task will help defer the execution same with this one:

is_xmlfile = file[0] == '...'

So I have to move the logic of checking what type of file it is to an own task?

Yes exactly because everything in the apply_map is a task. So basically apply_map just unpacks all of the tasks inside it and adds them all to the Flow block so you can think that it’s the same thing that you can’t have non tasks in the Flow block (for the most part)

Okay thank you! It works now, but takes a very long time (because there are about 30 files per zip file it runs a lot of tasks) Do you think my flow design for this use case is a good idea or would you handle it differently?

Looks fine to me right? I don’t know what more you can optimize. What is a very long time? 1 hour?

Just 3 minutes 😄 But without indiviual tasks it was around 20 seconds

Ahh ok then Prefect is the bottleneck. Each task has at least 3 API calls to update state (Pending -> Submitted -> Running -> Success), which is where the overhead comes from.

@Marius Haberstock catching up here: I didn't see any executor attached to your flow. If you want to optimize performance of this flow, you can run your tasks in parallel by attaching e.g. LocalDaskExecutor flow.executor = LocalDaskExecutor()