Hey all Im running a flow using VertexRun It has multiple st Prefect Community #ask-community

Hey all, Im running a flow using VertexRun - It ha...

Dekel R

01/12/2022, 10:07 AM

Hey all, Im running a flow using VertexRun - It has multiple steps and it runs perfectly. Once a week at a specific time I’d like this flow to invoke another flow that will train a model. I used this code -

Copy code

if pendulum.today('America/New_York').weekday() == 2:  # Monday is 0 so Wednesday is 2
        x_flow = create_flow_run(flow_name=PREFECT_TRAIN_FLOW_NAME, project_name=PREFECT_TRAIN_PROJECT_NAME)
        wait_for_flow_a = wait_for_flow_run(x_flow, raise_final_state=True)

This code is of course inside my “with Flow….” code block. Now when running this code alone (in a dummy flow) - it works and x_flow gets invoked. But when running this code in my real flow, after some other tasks - nothing happens. I cannot even see the task of “wait_for_flow” in prefect cloud(flow -> tasks tab) - seems like its getting ignored. What am I missing here? Thanks

Anna Geller

01/12/2022, 10:19 AM

Conditional logic with if/else statements should not be used in the Flow block in Prefect 1.0. This will change in Orion, but for now you need to use the case task, because the Flow block only defines a DAG, which gets execute when flow run is created.

Anna Geller

01/12/2022, 10:25 AM

btw, you can entirely avoid this if/else block if you attach this schedule:

Copy code

import pendulum
from prefect.schedules import CronSchedule


schedule = CronSchedule(
    cron="0 9 * * WED", start_date=pendulum.datetime(2022, 1, 12, 0, 0, tz="America/New_York")
)

for sched in schedule.next(10):
    print(sched)

it will run your flow every Wednesday your time zone at 9 AM (just example time)

Dekel R

01/12/2022, 10:28 AM

Yeah its not the case here - this flow is running every day and its calling another flow just on Wednesday. I will use case tasks for that, thanks!!

6 Views

Open in Slack

Previous Next