https://prefect.io logo
d

Donnchadh McAuliffe

02/18/2022, 4:08 PM
Hi there. I am new to Prefect Orion. I am trying to schedule a flow run on my local Prefect server following the example POST request provided in the documentation. Here is my request:
Copy code
POST: <http://localhost:4200/api/flow_runs/>

PAYLOAD:
{
  "name": "my_scheduled_flow",
  "deployment_id": "0ff88fc8-a71c-4b71-b8dd-078d6a36fb39",
  "flow_id": "94374bea-70ae-4e49-8988-e9077daa280a",
  "flow_version": "1.0",
  "state": {
    "type": "SCHEDULED",
    "name": "my_scheduled_flow",
    "message": "a message",
    "state_details": {
        "scheduled_time": "2022-02-18T16:10:49.005Z"
    }
  }
}
The flow run then appears in the server UI (as yellow) but never gets ran at the correct time. Does anyone have any idea on this? Am I missing an important part of the payload.
k

Kevin Kho

02/18/2022, 4:22 PM
Hi @Donnchadh McAuliffe, I’ll ask one of the Orion engineers
d

Donnchadh McAuliffe

02/18/2022, 4:24 PM
thank you @Kevin Kho
I managed to get a scheduled flow run to work in the end with the same format of payload - I had to reset the database, pip uninstall prefect, reinstall it, initiate the orion server and then send the payload. Thanks anyway!
k

Kevin Kho

02/18/2022, 4:37 PM
Sounds good. Thanks for the info!
d

Donnchadh McAuliffe

02/18/2022, 5:13 PM
if possible @Kevin Kho could you provide an example REST API payload for creating an interval scheduled flow run? https://orion-docs.prefect.io/api-ref/rest-api/#/Flow%20Runs/create_flow_run_flow_runs__post For example, a flow run that should start at
2022-02-18T17:11:43.254Z
and then every 20 mins until the next day (
2022-02-18T18:11:43.254Z
) Thank you!
k

Kevin Kho

02/18/2022, 5:15 PM
It’s really not recommended that you make the API Payload for that yourself because there are a bunch of steps needed to construct it, which is why the deployment interface exists
The code path is pretty long to replicate
d

Donnchadh McAuliffe

02/21/2022, 11:56 AM
apologies @Kevin Kho I should've clarified what I'm trying to do. Say I have already deployed a flow to my prefect server - what would the api request look like to run that flow with an interval schedule? Where do I add the start time/end time and run time interval to the payload?
Copy code
POST: <http://localhost:4200/api/flow_runs/>

PAYLOAD:
{
  "name": "my_scheduled_flow",
  "deployment_id": "0ff88fc8-a71c-4b71-b8dd-078d6a36fb39",
  "flow_id": "94374bea-70ae-4e49-8988-e9077daa280a",
  "flow_version": "1.0",
  "state": {
    "type": "SCHEDULED",
    "name": "my_scheduled_flow",
    "message": "a message",
    "state_details": {
        "scheduled_time": "2022-02-18T16:10:49.005Z"
    }
  }
}
a

Anna Geller

02/21/2022, 12:00 PM
You can attach a schedule to your DeploymentSpec:
Copy code
from prefect import flow

@flow
def hello_world(name="world"):
    print(f"Hello {name}!")

from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import IntervalSchedule
from datetime import timedelta

DeploymentSpec(
    flow=hello_world,
    name="hello-world-daily",
    schedule=IntervalSchedule(interval=timedelta(days=1)),
    tags=["earth"],
)

DeploymentSpec(
    flow=hello_world,
    name="hello-orion",
    schedule=IntervalSchedule(interval=timedelta(weeks=1)),
    parameters={"name": "Orion"},
    tags=["marvin"],
)
d

Donnchadh McAuliffe

02/21/2022, 12:08 PM
thank you, but say I don't want to specify the schedule when deploying the flow and I just want to deploy a generic flow and then run it using an API request. I have achieved this with scheduling a one off run as can be seen above, but how do I modify my payload to specify an interval schedule flow run?
how do I add an end time and time interval to this section of the payload to run an already deployed flow?
Copy code
"state_details": {
        "scheduled_time": "2022-02-18T16:10:49.005Z"
    }
a

Anna Geller

02/21/2022, 12:17 PM
I believe you can either: • deploy your flow to run on schedule as shown above with
DeploymentSpec
- the flow runs then get created via schedule • or trigger the flow run ad hoc via an API call as you did I think creating deployments and attaching schedules via an API call is not yet tested/documented so I would wait with that a bit.
d

Donnchadh McAuliffe

02/21/2022, 12:23 PM
ok thank you @Anna Geller, so for now the API call method only works for a one off schedule by the looks of it. If possible can you confirm that triggering an interval schedule flow run via an API call is on the roadmap for Prefect Orion official release? Really appreciate your help with this!
a

Anna Geller

02/21/2022, 12:34 PM
Sorry, I can't make any promises about the roadmap. But since this seems important to you I checked the API and it seems there is a route you can use to schedule a run for a deployment https://orion-docs.prefect.io/api-ref/rest-api/#/Deployments/schedule_deployment_deployments__id__schedule_post
d

Donnchadh McAuliffe

02/21/2022, 5:04 PM
fair enough, thank you @Anna Geller. The payload provided in the doc doesn't actually work I'm afraid:
Copy code
POST: <http://localhost:4200/api/deployments/{deployment_id}/schedule>

PAYLOAD: {
  "start_time": "2022-02-21T17:00:00.000Z",
  "end_time": "2022-02-21T17:02:00.000Z",
  "max_runs": 4
}
it just returns
null, 200
and subsequently doesn't run at the specified times. I'm looking through the orion docs for schedules here, how do I specify the start and end time for the interval schedule of a deployment?
Copy code
from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import IntervalSchedule
from datetime import timedelta

DeploymentSpec(
    flow_location="/Developer/workflows/my_flow.py",
    name="my-first-deployment",
    parameters={"nums": [1, 2, 3, 4]}, 
    schedule=IntervalSchedule(interval=timedelta(minutes=15)),
)
a

Anna Geller

02/21/2022, 7:46 PM
I looked at it a bit more and I would encourage you to wait until you use this feature anywhere since it’s not yet fully tested/documented. This API endpoint is not yet meant to set schedules via API and it doesn’t work as you would like. Also, the API routes and database models are still under active development so even if this would work, there is no guarantee that this won’t change. In contrast, defining a schedule via a
DeploymentSpec
has been stable for many weeks now and I would encourage you to set schedule only via DeploymentSpec for now.
d

Donnchadh McAuliffe

02/21/2022, 8:43 PM
sounds good @Anna Geller, just final question from me regarding your recommendation "I would encourage you to set schedule only via DeploymentSpec for now." How do I add the start and end times to an interval schedule in the below python example? Its not specified in your schedule docs.
Copy code
from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import IntervalSchedule
from datetime import timedelta

DeploymentSpec(
    flow_location="/Developer/workflows/my_flow.py",
    name="my-first-deployment",
    parameters={"nums": [1, 2, 3, 4]}, 
    schedule=IntervalSchedule(interval=timedelta(minutes=15)),
)
a

Anna Geller

02/21/2022, 8:54 PM
That’s actually a good point, I will open an issue internally to discuss this since in Prefect 1.0 we do have start and end dates for both Cron and Interval schedules. Thanks for raising that
d

Donnchadh McAuliffe

02/21/2022, 8:55 PM
much appreciated!
z

Zanie

02/21/2022, 11:16 PM
Just out of curiosity, what’s your use-case for start and end dates on a schedule?
d

Donnchadh McAuliffe

02/22/2022, 9:25 AM
say you had a flow to collect data throughout an event that starts at a certain time and ends at a certain time. For example if you wanted to collect sports data intermittently from an API (say every 2mins) throughout a soccer match that takes place on a Saturday - you don't want to have to manually run the interval schedule at the start of the match and then manually stop it after 90mins (its the weekend and you're not on your laptop the entire time, instead on the Friday night you just want to "schedule" the interval schedule to begin at the match start time, run every 2mins and end after 90mins).
@Anna Geller is there a github issue I could follow regarding the missing capability of adding start and end times to interval schedule on Orion?
a

Anna Geller

02/22/2022, 12:51 PM
Unfortunately, Orion repo is currently private and I opened an issue there. But I'll do my best to keep you up to date. There's a lot to do regarding Orion, so it may take some time
d

Donnchadh McAuliffe

02/22/2022, 12:54 PM
thats fine, thanks again for all your help!
👍 1
Also, @Anna Geller or @Zanie if I could bother ye with another scheduling question. The docs say you can only create 100 flow runs for a deployment that contains an interval schedule using the default settings. How do I change the settings in order to make more than 100 flow runs for a schedule? For example if I want to run the flow with an interval schedule of every hour for 2 weeks, a total of 336 flow runs.
a

Anna Geller

02/24/2022, 11:48 AM
Your scheduled runs will be created as expected. The 100 flow runs is just an implementation detail of the scheduler services which tells you that 100 runs will be created immediately, but the next runs for your 2 week-period will be created later by the scheduler. E.g. in Prefect 1.0, the scheduler creates only 10 runs in advance, but all scheduled runs will be eventually created as they are getting closer to their expected scheduled start time.
d

Donnchadh McAuliffe

02/24/2022, 11:51 AM
ya thats what I thought but just wanted to confirm with ye. Thanks for answering all these questions, its helping us rule out any potential limitations we might have if we commit to Orion.
👍 1
Hey @Anna Geller. Just wondering if the
end_time
and
start_time
has been added to Orion?
Copy code
from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import IntervalSchedule
from datetime import timedelta

DeploymentSpec(
    flow_location="/Developer/workflows/my_flow.py",
    name="my-first-deployment",
    parameters={"nums": [1, 2, 3, 4]}, 
    schedule=IntervalSchedule(interval=timedelta(minutes=15)),
)
a

Anna Geller

03/30/2022, 4:32 PM
Not yet, here is an issue so that you can track the status https://github.com/PrefectHQ/prefect/issues/5620
d

Donnchadh McAuliffe

03/30/2022, 4:44 PM
thanks so much
👍 1
3 Views