Satheesh K
04/15/2021, 2:38 PMintermediate_result = 0
with Flow("flow") as flow:
param1 = Parameter("list1")
mapped_list = create_map(param1)
results = task1.map(mapped_list)
intermediate_result = results[1]
results2 = task2(results[1])
Kevin Kho
@task
def pull_first_element(results):
return results[1]
Kevin Kho
intermediate_result = pull_first_element(results)
Satheesh K
04/15/2021, 2:43 PMKevin Kho
Satheesh K
04/15/2021, 2:46 PMSatheesh K
04/15/2021, 2:46 PMSatheesh K
04/15/2021, 2:46 PMKevin Kho
import prefect
from prefect import Flow, task, unmapped
@task
def get_numbers():
return [1,2,3]
@task
def first(x):
return x[0]
@task
def my_log(first):
logger = prefect.context.get("logger")
<http://logger.info|logger.info>(f"Pulled the first value {first}")
return first
with Flow('test-flow') as flow:
n = get_numbers()
first = first(n)
my_log(first)
flow.run()
Satheesh K
04/15/2021, 2:55 PMmy_log
Kevin Kho
Pulled the first value 1
Satheesh K
04/15/2021, 2:58 PMinter_res
with 1.Kevin Kho
Satheesh K
04/15/2021, 2:59 PMTask
instance.Kevin Kho
print
inside the Flow will not work because print is executed in build time but the values are populated in runtime.Kevin Kho
Kevin Kho
inter_res
will be assigned during build time but the value of my_log(first)
will still just be a Task. You can’t get inter_res
to be 1 inside a FlowSatheesh K
04/15/2021, 3:02 PMKevin Kho
Kevin Kho
Satheesh K
04/15/2021, 3:04 PMSatheesh K
04/15/2021, 3:08 PMSatheesh K
04/15/2021, 4:30 PMKevin Kho