How to map over a list of inputs and map over the results? #3287

gryBox · 2020-09-09T15:49:26Z

gryBox
Sep 9, 2020

Hi - Below is a Flow that returns the expected results when the input is a single number .i.e 3. However I am at a loss on how to execute the same logic with a list of inputs, input_num_lst= [3, 2] where the result would be five mapped tasks. I did try apply_map but to no avail. -- Thanks

Working example with input NOT a list

returns 3 mapped Items

from prefect import Flow, task, apply_map, Parameter, unmapped
import random

@task
def list_maker(input_num):
  print(input_num)
  lst = [random.randint(1, 10) for i in range(input_num)]
  print(lst)
  return lst
  
@task
def expensive_func(num_from_random_lst, num_from_input):
  return num_from_random_lst*num_from_input

with Flow("map the map") as flow:
  input_num_lst = Parameter("input_num_lst", default=3)

  random_list = list_maker(input_num)
  expensive_func = expensive_func.map(random_list, unmapped(input_num_lst))

@marvin-robot

Answered by jcrist

Sep 10, 2020

Sorry, I'm left guessing at what you actually want here. Here's another attempt:

from prefect import Flow, task, apply_map, Parameter, unmapped, flatten
import random


@task
def list_maker(input_num):
    return [(random.randint(1, 10), input_num) for i in range(input_num)]


@task
def expensive_func(input):
    num_from_random_lst, num_from_input = input
    return num_from_random_lst * num_from_input


with Flow("map the map") as flow:
    input_num_lst = Parameter("input_num_lst", default=[2, 3])

    random_list = list_maker.map(input_num_lst)
    output = expensive_func.map(flatten(random_list))

state = flow.run()
print(state.result[output].result)

$ python test.py
[2020-09-10 17:4…

View full answer

jcrist · 2020-09-09T22:55:20Z

jcrist
Sep 9, 2020

Hi @gryBox, the above doesn't run clean for me. Rather than guess what you want and where the typo is, can you fix your provided code to run cleanly? After that, which input(s) to what task(s) do you want to change from a scalar value to a mapped value?

from prefect import Flow, task, apply_map, Parameter, unmapped
import random


@task
def list_maker(input_num):
    print(input_num)
    lst = [random.randint(1, 10) for i in range(input_num)]
    print(lst)
    return lst


@task
def expensive_func(num_from_random_lst, num_from_input):
    return num_from_random_lst * num_from_input


with Flow("map the map") as flow:
    input_num_lst = Parameter("input_num_lst", default=3)

    random_list = list_maker(input_num)
    expensive_func = expensive_func.map(random_list, unmapped(input_num_lst))


flow.run()

$ python test.py
Traceback (most recent call last):
  File "test.py", line 21, in <module>
    random_list = list_maker(input_num)
NameError: name 'input_num' is not defined

2 replies

gryBox Sep 10, 2020
Author

@jcrist Sorry about that and thanks for the help. I fixed the example. This is how the run looks when the input Paramater is 3.

Now if the list_maker task is changed to a mapped task. I get this. What I magically expect is five terminal tasks.

Code with mapped:

from prefect import Flow, task, apply_map, Parameter, unmapped
import random

@task
def list_maker(input_num):
  print(input_num)
  lst = [random.randint(1, 10) for i in range(input_num)]
  print(lst)
  return lst
  
@task
def expensive_func(num_from_random_lst, num_from_input):
  return num_from_random_lst*num_from_input

with Flow("map the map") as flow:
  input_num_lst = Parameter("input_num_lst", default=[2,3])

  random_list = list_maker.map(input_num_lst)
  expensive_func = expensive_func.map(random_list, unmapped(input_num_lst))

jcrist Sep 10, 2020

I think what you're looking for is prefect.flatten. Mapping over a function that returns a list will result in a list-of-lists. You then want to "flatten" this list-of-lists into a single list you can map expensive_func over. The following does that:

from prefect import Flow, task, Parameter, unmapped, flatten
import random


@task
def list_maker(input_num):
    return [random.randint(1, 10) for i in range(input_num)]


@task
def expensive_func(num_from_random_lst, num_from_input):
    return num_from_random_lst * num_from_input


with Flow("map the map") as flow:
    input_num_lst = Parameter("input_num_lst", default=[2, 3])

    random_list = list_maker.map(input_num_lst)
    expensive_func = expensive_func.map(flatten(random_list), unmapped(input_num_lst))

See https://docs.prefect.io/core/concepts/mapping.html#flat-mapping for more info.

gryBox · 2020-09-10T16:54:36Z

gryBox
Sep 10, 2020
Author

@jcrist Using your flow produces strange results both visually and as output. Between this roadblock and another issue I've opened up Multiple flows cannot be used with the same case statement I think it's time for me to re-architect the flow, so I don't run into these contortions?

Results

[[2020-09-10 16:40:55] INFO - prefect.FlowRunner | Beginning Flow run for 'map the map'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'input_num_lst': Starting task run...
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'input_num_lst': finished task run for task with final state: 'Success'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'list_maker': Starting task run...
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'list_maker': finished task run for task with final state: 'Mapped'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'list_maker[0]': Starting task run...
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | 2
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | [7, 4]
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'list_maker[0]': finished task run for task with final state: 'Success'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'list_maker[1]': Starting task run...
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | 3
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | [5, 5, 10]
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'list_maker[1]': finished task run for task with final state: 'Success'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func': Starting task run...
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func': finished task run for task with final state: 'Mapped'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func[0]': Starting task run...
7 [2, 3]
[2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3]
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func[0]': finished task run for task with final state: 'Success'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func[1]': Starting task run...
4 [2, 3]
[2, 3, 2, 3, 2, 3, 2, 3]
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func[1]': finished task run for task with final state: 'Success'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func[2]': Starting task run...
5 [2, 3]
[2, 3, 2, 3, 2, 3, 2, 3, 2, 3]
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func[2]': finished task run for task with final state: 'Success'
[2020-09-10 16:40:55] INFO - prefect.TaskRunner | Task 'expensive_func[3]': Starting task run...
5 [2, 3]
[2, 3, 2, 3, 2, 3, 2, 3, 2, 3]
[2020-09-10 16:40:56] INFO - prefect.TaskRunner | Task 'expensive_func[3]': finished task run for task with final state: 'Success'
[2020-09-10 16:40:56] INFO - prefect.TaskRunner | Task 'expensive_func[4]': Starting task run...
10 [2, 3]
[2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3]
[2020-09-10 16:40:56] INFO - prefect.TaskRunner | Task 'expensive_func[4]': finished task run for task with final state: 'Success'
[2020-09-10 16:40:56] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
[[2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3], [2, 3, 2, 3, 2, 3, 2, 3], [2, 3, 2, 3, 2, 3, 2, 3, 2, 3], [2, 3, 2, 3, 2, 3, 2, 3, 2, 3], [2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3, 2, 3]]]([url](url))

0 replies

jcrist · 2020-09-10T17:38:27Z

jcrist
Sep 10, 2020

Sorry, I'm left guessing at what you actually want here. Here's another attempt:

from prefect import Flow, task, apply_map, Parameter, unmapped, flatten
import random


@task
def list_maker(input_num):
    return [(random.randint(1, 10), input_num) for i in range(input_num)]


@task
def expensive_func(input):
    num_from_random_lst, num_from_input = input
    return num_from_random_lst * num_from_input


with Flow("map the map") as flow:
    input_num_lst = Parameter("input_num_lst", default=[2, 3])

    random_list = list_maker.map(input_num_lst)
    output = expensive_func.map(flatten(random_list))

state = flow.run()
print(state.result[output].result)

$ python test.py
[2020-09-10 17:40:04] INFO - prefect.FlowRunner | Beginning Flow run for 'map the map'
[2020-09-10 17:40:04] DEBUG - prefect.FlowRunner | Using executor type LocalExecutor
[2020-09-10 17:40:04] DEBUG - prefect.FlowRunner | Flow 'map the map': Handling state change from Scheduled to Running
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'input_num_lst': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'input_num_lst': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'input_num_lst': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'input_num_lst': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'input_num_lst': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'list_maker': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'list_maker': Handling state change from Pending to Mapped
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'list_maker': finished task run for task with final state: 'Mapped'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'list_maker[0]': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'list_maker[0]': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'list_maker[0]': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'list_maker[0]': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'list_maker[0]': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'list_maker[1]': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'list_maker[1]': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'list_maker[1]': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'list_maker[1]': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'list_maker[1]': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func': Handling state change from Pending to Mapped
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func': finished task run for task with final state: 'Mapped'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[0]': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[0]': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[0]': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[0]': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[0]': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[1]': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[1]': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[1]': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[1]': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[1]': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[2]': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[2]': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[2]': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[2]': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[2]': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[3]': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[3]': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[3]': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[3]': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[3]': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[4]': Starting task run...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[4]': Handling state change from Pending to Running
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[4]': Calling task.run() method...
[2020-09-10 17:40:04] DEBUG - prefect.TaskRunner | Task 'expensive_func[4]': Handling state change from Running to Success
[2020-09-10 17:40:04] INFO - prefect.TaskRunner | Task 'expensive_func[4]': finished task run for task with final state: 'Success'
[2020-09-10 17:40:04] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
[2020-09-10 17:40:04] DEBUG - prefect.FlowRunner | Flow 'map the map': Handling state change from Running to Success
[20, 16, 24, 3, 24]

expensive_func gets called with a tuple of:

A random value generated in list_maker
Your unmapped value from input_num_lst (either 2 or 3 depending)

It gets called 5 times (2 + 3), and output is a list of the results. If this is not what you actually want the flow to do, you'll need to better describe your expectations here.

0 replies

gryBox · 2020-09-11T05:54:09Z

gryBox
Sep 11, 2020
Author

@jcrist Exactly the output I was looking for.

Does this visualization raise any red flags?
Would memory be an issue if the inputs were large. i.e spacy nlp task results?

Again I appreciate your help.

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

How to map over a list of inputs and map over the results? #3287

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Replies: 4 comments 2 replies

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

Select a reply

How to map over a list of inputs and map over the results? #3287

gryBox Sep 9, 2020

Working example with input NOT a list

Replies: 4 comments · 2 replies

jcrist Sep 9, 2020

gryBox Sep 10, 2020 Author

jcrist Sep 10, 2020

gryBox Sep 10, 2020 Author

Results

jcrist Sep 10, 2020

gryBox Sep 11, 2020 Author

gryBox
Sep 9, 2020

Replies: 4 comments 2 replies

jcrist
Sep 9, 2020

gryBox Sep 10, 2020
Author

gryBox
Sep 10, 2020
Author

jcrist
Sep 10, 2020

gryBox
Sep 11, 2020
Author