Hi I have a question about `apply map` Details following in

Question

Hi I have a question about `apply map` Details following in the thread slightly smiling face

Marius Haberstock · Answer

I have a flow where I first download a zip file unpack it and then further process some files of the zip content I now want to split these steps into individual tasks My idea was to use a task for the downloading and unpacking part which returns a list of file names Then I would use a `iterate files` task to decide for each file if I call task `process file a` or task `process file b` I tried using `apply map` based on the example of the documentation <https docs prefect io core concepts mapping html complex mapped pipelines> However the `merge` function does not seem to work inside a task This is my current code `with Flow as flow ` `zip filename = Parameter zip filename ` `files = download zipfile zip filename ` `infos = apply map iterate files file=files zip filename=unmapped zip filename ` `upload file map info=infos ` ` task log stdout=True ` `def iterate files file zip filename ` `is recfile = file 0 endswith ` `with case is recfile True ` `result 1 = parse recfile run file zip filename ` `is xmlfile = file 0 == ` `with case is xmlfile True ` `result 2 = process xml file run file zip filename ` `return merge result 1 result 2 run ` Does anyone have a idea how to solve this slightly smiling face

Kevin Kho · Answer

I think it works The example in docs uses it <https docs prefect io core concepts mapping html complex mapped pipelines|here> Maybe don t call the `run `

Kevin Kho · Answer

There are a couple things off 1 Apply map is on a function not a task 2 all the contents of the apply map have to be tasks 3 You don t call the run method in it

Marius Haberstock · Answer

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 lt Task parse recfile gt 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 ``

Kevin Kho · Answer

That is because you have the task decorator still above the `iterate files`

Marius Haberstock · Answer

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 lt module gt ` `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 `

Kevin Kho · Answer

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 == ```

Marius Haberstock · Answer

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

Kevin Kho · Answer

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

Marius Haberstock · Answer

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

Kevin Kho · Answer

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

Marius Haberstock · Answer

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

Kevin Kho · Answer

Ahh ok then Prefect is the bottleneck Each task has at least 3 API calls to update state Pending gt Submitted gt Running gt Success which is where the overhead comes from

Kevin Kho · Answer

So upto you if you want to combine it to reduce overhead

Anna Geller · Answer

< 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