import IPython
from prefect import task, Flow, Parameter


my_list = [1, 3, 4]


@task
def map1_fn(item):
    new_item = [item+1, item+1]
    print("map1_fn: {}".format(item))
    return new_item


@task
def map2_fn(item):
    # expecting item = 2 but got [2, 2] etc
    print("map2_fn: {}".format(item))
    return item


with Flow("workflow") as flow:
    my_other_list = map1_fn.map(item=my_list)
    my_another_list = map2_fn.map(item=my_other_list)

flow.run()

my_other_list

[[2, 2], [4, 4], [5, 5]]

map2_fn

item

map2_fn

def my_reduce_task(my_list):
    ... # flatten list
    return flattened_list

flat_list = my_reduce_task(my_other_list)