Hello Community. I have a situation here. I want to run a flow as suggested by my mentor as we get an event. It processes payload . Currently we are using process pool to create process pool. Problem is we are getting high latencies. import json from pathlib import Path from prefect import flow from prefect.events.schemas.automations import Posture from prefect.events.schemas.deployment_triggers import DeploymentEventTrigger from prefect.events.schemas.events import ResourceSpecification # Create a basic flow with deployment Trigger to Run this flow as after certain Events. # Following Code uses minimum options to create a DeploymentEventTrigger. deployment_trigger = DeploymentEventTrigger( expect={"custom.event.*"}, # Match Resources to filter Deployments match=ResourceSpecification({'prefect.resource.id':["prefect.custom.event.*"]}), posture=Posture.Reactive, # Evaluate the trigger separately for each distinct value of these label for_each={"prefect.resource.id"}, # Set threshold to 1 and with in to 0 to trigger automation for every event threshold=1, within=0, # Sent event payload to flow as

payload

argument parameters={"payload": {"data":"{{ event.payload|tojson }}", "occured":"{{ event.occurred.timestamp() }}"}} ) @flow(log_prints=True) def data_pipeline(payload): data = json.loads(payload['data']) n_o_w, e_p_s, n_o_m=data['data']['number_of_workers'], data['data']['events_per_second'], data['data']['number_of_minutes'] with open(f'some_{n_o_w}_{e_p_s}_{n_o_m}.csv', 'a') as f: print(f'{data['event']}', file=f, flush=True) return 1 # Create Deployment from Source if name == "__main__": data_pipeline.from_source( source=str(Path(file).parent), entrypoint="basic_flow.py:data_pipeline", ).deploy(name="ddata piplevm", work_pool_name='process-worker-pool', description='None', triggers=[deployment_trigger] ) above is flow defination. We are testing by manually emitting say 10 events per second . The latecy data is attached as csv. We emiited 5 events per second for 2 minutes while 10 workers actively polling for flows. Execution envoirnment: OS: windows 10 Prefect version:3.1.7 (prefect local server with sqlite database) Processor: 11th Gen Intel(R) Core(TM) i5-1135G7 @ 2.40GHz 2.42 GHz 8 logical cores Memory: 16.0 GB (15.8 GB usable) Python:3.12.8 How can we address this problem in efficient way to reduce latency? Is there a solution?

hi @Krishna Suryawanshi - can you please place the code snippet in the thread? also im not clear on what latency you're describing, is it the amount of time after the event is created that the flow run gets triggered?

Thank you for replying. Yes sir 2 latencies 1. amount of time flow run is created after event is occured 2. amount of time flow run is started after flow run is created basic_flow.py contains a minimal flow event_emitter.py contains code that emits events based on user defined numbers

with the latencies, what are your expectations and observations?

Well my mentor asked if can we reach as real time as possible ,possibly less than 500 ms.

are you using prefect cloud or prefect open source?

prefect open source

on open source, the events should be propagated quickly enough for that, but the worker doesn't check for scheduled runs that often. so you could set the

PREFECT_WORKER_QUERY_SECONDS=0.5

, but note that will 20x how often the worker asks the server for work (defaults to 10 seconds) if you want full control you can use the websocket client directly and not use a deployment, worker or trigger and call your flow functions when you see the prerequisite events but that would be more work (since you'd re-implement triggers)

Thank you, I will try to work with this. Thank you for your help. If there is anything will post on community. Thank you very much.

the worker doesn't check for scheduled runs that often. so you could set the

PREFECT_WORKER_QUERY_SECONDS=0.5

,

i think this is likely the most relevant piece of info. the worker polls the server to find scheduled runs, but by default it only checks every 10 seconds have you changed this setting?

Yes Already did but didn't find any improvement.

thank you! i shouldnt have recommended such a low value in retrospect - OP had mentioned

Well my mentor asked if can we reach as real time as possible ,possibly less than 500 ms.

and i thought it was a

float

- my bad thanks for following up

Hello @Nate @Zarinah can you help me reduce latency between when event was occurred and event was created i.e. second last column in https://prefect-community.slack.com/files/U086WDB0C81/F08637LQ7FZ/data_10_5_2.csv?origin_team=TL09B008Y&origin_channel=CKNSX5WG3 in this file in case of multiple events per second.

> @Krishna Suryawanshi, By changing

PREFECT_RUNNER_POLL_FREQUENCY

, which is different than

PREFECT_WORKER_QUERY_SECONDS

I was able to reduce the schedule to start latency. There is some time in between where Prefect seems to have some amount of overhead in starting up the job. It shows in the UI as "Pending", I couldn't figure out how to get that lower than 2s. I've been kicking things off by manually running the flow. In that case it's fairly instant to start. I don't believe prefect is meant to be real-time low latency. I'm new to Prefect, so correct me if I'm wrong. My plan for flows I really want to kick off instantly was to create a docker that runs something like FastAPI to kick off the job vs using the Prefect dashboard to kick off. @Krishna Suryawanshi, looking at your table, I'm wondering if the issue has to do with resources or number of workers available to run. Looks like you kick off a bunch of jobs at the same time. I'm guessing some of them need to wait for previous ones to finish? Like I said, I'm new, that's just my guess.

unfortunately, to some extent there is a latency out of your control which is "how long does it take for the event travel through the message bus in prefect cloud?" if you use open source, you'll notice that events are propagated nearly instantly, but as cloud serves all users, there is a non zero latency for events to come through (that we are always working on decreasing!)

Hello @Nate @Zarinah I attached little more csv file, this time with different number of workers. Can you please give it a try and help me? Settings are this: PREFECT_RUNNER_POLL_FREQUENCY='1' PREFECT_WORKER_QUERY_SECONDS='0.5' These three are files such data_(no of_worker}__{events__per_second}__{number_of_minutes}.csv First column= event number Second Column = flow_run created - event_occured Third Column = flow run started - flow run created Values in third column is not linear but lies between some interval as we can see from attachements. As the second column is almost linear function of event emitted even in case of different number of workers . Can we parallelize something on server side such that we can say that maximum value of second column lies between some known interval as in case of third column