Hi all! Does anyone have a best practice for a “pr...
# ask-community
a
Hi all! Does anyone have a best practice for a “prefect-equivalent” list comprehension on the result of a mapped task? Trying:
Copy code
from prefect import Flow, task

@task()
def some_task(x):
    return x+1

some_list = [1, 2, 3, 4]

with Flow("list_comp") as list_comp_flow:
    mapped_result = some_task.map(some_list)
    list_comp = [result+1 for result in mapped_result]
results in the following error:
Copy code
TypeError: Task is not iterable. If your task returns multiple results, pass `nout` to the task decorator/constructor, or provide a `Tuple` return-type annotation to your task.
Which makes sense of course, since while building the flow,
mapped_result
is still a
Task
, not a
list
.
s
Wouldnt this be
map(some_task, some_list)
instead of the other way around?
EDIT: Apologies, I see prefect tasks implement a map function, Ill be following this too. Their docs here https://docs.prefect.io/core/concepts/mapping.html#prefect-approach seem to imply that you should have a function which does the result+1 so that it can be executed as a task at runtime
a
Thanks Samuel! That is indeed a proper solution, and in line with the Prefect mindset, and perhaps the best way to go about it. I find myself reluctant to write a task for each “small” list comp, but that might just be my own habits, and I need to get used to this 🙂
s
I was with you in the same mindset, having come from airflow (where waiting for a task to be picked up could take minutes and thus lots of small tasks slowed things down a ton). With a dask executor running my taks though, its been super quick, this flow of mine with dozens and dozens of tasks executres in a couple of seconds:
a
Here is my solution, for what it’s worth:
Copy code
from prefect import Flow, task
@task()
def some_task(x):
    return x+1

@task
def list_comprehension(original_list, lambda_func):
    return [lambda_func(x) for x in original_list]

some_list = [1, 2, 3, 4]

with Flow("list_comp") as list_comp_flow:
    mapped_result = some_task.map(some_list)
    list_comp = list_comprehension(mapped_result, lambda x: x+2)
Still creates a new task, but I don’t have to write a specific task every time I want a quick list comp
👍 1
k
You can also just pass the lambda inside a
task()
call since the decorator is just syntactic sugar for the same thing
Copy code
from prefect import Flow, task
@task()
def some_task(x):
    return x+1

some_list = [1, 2, 3, 4]
with Flow("list_comp") as list_comp_flow:
    mapped_result = some_task.map(some_list)
    list_comp = task(lambda x: x+1).map(mapped_result)
🙏 1
🙌 1
But I wouldn’t say it’s recommended because you just a get generic “lambda…” in the logs
a
Not necessarily worse then my generic “list_comprehension” in the logs…
k
Note the two will have different flow graphs. The one you had makes
list_comprehension
one task while this makes another
map
operation.
1