Local Prefect project won't print/log results

user_1377288 0
Viewed 562

I've been trying for a while to run parametrised and connected flows.

Basically, there's a task that returns a dictionary of parameters (make_poem), and a flow that prints something based on the parameters (declaim_poem). There's also the following parent-flow:

flow_run = StartFlowRun(flow_name="declaim_flow", project_name=project_name)
with Flow("param_flow") as flow:
    flow_run.set_upstream(task(make_poem), key="parameters")

Which is than ran, and this is the result:

[2021-08-01 17:04:05+0200] INFO - prefect.FlowRunner | Beginning Flow run for 'param_flow'
[2021-08-01 17:04:05+0200] INFO - prefect.FlowRunner | Beginning Flow run for 'param_flow'
[2021-08-01 17:04:05+0200] INFO - prefect.FlowRunner | Beginning Flow run for 'param_flow'
[2021-08-01 17:04:05+0200] INFO - prefect.FlowRunner | Beginning Flow run for 'param_flow'
[2021-08-01 17:04:05+0200] INFO - prefect.TaskRunner | Task 'make_poem': Starting task run...
[2021-08-01 17:04:05+0200] INFO - prefect.TaskRunner | Task 'make_poem': Finished task run for task with final state: 'Success'
[2021-08-01 17:04:05+0200] INFO - prefect.TaskRunner | Task 'Flow declaim_flow': Starting task run...
[2021-08-01 17:04:06+0200] INFO - prefect.Flow declaim_flow | Flow Run: < cloud link >
[2021-08-01 17:04:06+0200] INFO - prefect.TaskRunner | Task 'Flow declaim_flow': Finished task run for task with final state: 'Success'
[2021-08-01 17:04:06+0200] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded

Even though it's a success result, I can't get the expected output from declaim_poem (as implied, a poem) to print out on the screen. I'm using a logger instantiated inside the declaim_poem task, not flow, as per this page.

In the Prefect Cloud link provided, I also can't find the poem, only successful results from all the tasks from all flows.

Any ideas on what I'm doing wrong?

prefect python-3.x python
1 Answers

Try this, different API but do the same thing as StartRunFlow with outputting the sub_flow logs.

from prefect import Flow

from prefect.tasks.prefect.flow_run import create_flow_run, wait_for_flow_run

with Flow("parent-flow") as flow:

    flow_1_run_id = create_flow_run(flow_name="sub_flow_1")
    wait_for_flow_run(flow_1_run_id, stream_logs=True)

    flow_2_run_id = create_flow_run(flow_name="sub_flow_2")
    wait_for_flow_run(flow_2_run_id, stream_logs=True)

References: https://www.prefect.io/blog/prefect-0-15-0-a-new-flow-run-experience/

user_49271720
Related