Hello, Ive tried Prefect today and I really love it as pipeline data from TMDB to my Supabase. But its is normal that my python script is very slow. I means I should update around 5000 items per days but its takes 3min for 100 items that pretty weird no ? Thx for help !

hi @Loup - welcome! there are many reasons why things could be moving slowly, especially if your work is IO / network bound could you show how you're utilizing prefect here and perhaps what your code is doing in general? prefect should not generally add that much latency

Thx for your help ! There is all my script :

When you want to run your script off of your local machine @Loup, this part of the getting started talks explain how to host it in a github repo: https://docs.prefect.io/latest/getting-started/quickstart/#step-4-choose-a-remote-infrastructure-location

The best option for me is to use Prefect Cloud because I don’t really wanna have an another VPS for this. But I’m getting an error with it. Also Prefect don’t give us the pricing of the Cloud, and I don’t know if we have a limit of storage space (because my script will download 2 files of 250mo at each execution) and machine specification either (cpu, ram,…). Ive tried to make a VPS using Docker with Prefect. It’s working but I need to have my VPS terminal open because if I leave the terminal (after started the pool worker), all my execution are in state mode (so I guess my worker isn’t reachable)

hi @Loup - even if you’re using prefect cloud, by default you will need to run a worker process that will listen for work to submit. we are in the process of rolling out managed execution where we do this for you - right now push pools are probably your best bet @Marvin please lookup push pools and also find that systemd linux worker article, also eli5 everything

It seems like there might be an issue with the link formatting that you've posted. To help you more effectively, could you please describe the problem you're encountering? Are you trying to access the Prefect documentation related to "push work pools"? If there's a specific issue or topic you need information on, please clarify, and I can assist you further.

and if you want to run your own worker on linux

Sure, let's break it down into simpler terms: Work Pools in Prefect 2.x: Work pools are like designated areas where flow runs wait to be performed. When you have a flow that's ready to run, it's sent to a specific work pool. Think of it as a bus stop where the flow run waits for a bus (the worker) to pick it up and take it to its destination (execution). Workers in Prefect 2.x: Workers are like the bus drivers for your flows. They check in with the work pool to see if there are any flow runs that need to be done. If they find any, they pick up the flow run and execute it. You don't need to keep your laptop open because these workers are running elsewhere, like in the cloud or on a server, doing the work automatically. Push Pools in Prefect 2.x: Push work pools are a special kind of bus stop that doesn't need a bus driver at all! Instead of waiting for a worker to pick up the flow run, the work pool directly sends the flow run to serverless infrastructure, such as AWS ECS Tasks or GCP Cloud Run Jobs. This means your flows can be run on-demand in these serverless services without needing a dedicated worker to start the run. Using work pools and workers, or push work pools, allows your flows to run automatically without needing your laptop to be on since cloud services or dedicated servers handle the execution for you.

@Nate thx Ive added a service to run autmatically my worker. Do I really need to use docker ? Or I can just create a local subprocess ? Because I dont know how can I add pip library to the docker image 😅

you don't need to use docker, the absolute easiest way if you just have one script is to use the

.serve

method and maybe put that up as a systemd process that revives itself

So I can just create a local subprocess worker ?

if you only have 1 script, then

serve

does the whole job of the Process worker - if you have

# my_new_deployment.py
from prefect import flow

@flow(log_prints=True)
def some_flow():
   pass

if __name__ == "__main__":
   some_flow.serve("my_new_deployment")

then you can do

python my_new_deployment.py

and that will start a long-lived process analogous to a process worker the only nice thing about a worker is that if you wanted to add another script (i.e. deployment) later to run in the same place, you'd just need to create the deployment (

prefect deployment somefile.py:somefunc -n new-deployment -p my-process-pool

) and the process worker would be able to pull it in (without having to change anything else) @Marvin concisely give a systemd config file to run the above script on linux as a service

To run your script as a service on a Linux system with

systemd

, you can create a service file (let's call it

my_flow_service.service

) with the following configuration:

ini
[Unit]
Description=My Flow Service
After=network.target

[Service]
Type=simple
User=<YourUsername>
WorkingDirectory=<YourScriptDirectory>
ExecStart=/usr/bin/python3 my_new_deployment.py
Restart=on-failure
RestartSec=5s

[Install]
WantedBy=multi-user.target

Replace

<YourUsername>

with your Linux username and

<YourScriptDirectory>

with the full path of the directory where your script

my_new_deployment.py

is located. You would then place this file in the

/etc/systemd/system/

directory. Enable and start the service using the following commands:

shell
sudo systemctl enable my_flow_service.service
sudo systemctl start my_flow_service.service

This will set up your script as a systemd service that starts automatically, runs continuously in the background, and restarts in case of failure.

where if you ended up wanting to go the worker route, you could instead run the worker process here

ExecStart=/usr/path/to/my/venv/prefect worker start -p my-process-pool

But with

.serve

we can't specify worker pool no ? And no I wanna have multiple script so maybe docker is better I guess, but dont know how add rule for adding pip library for my script

But with

.serve

we can't specify worker pool no ?

correct. you'd have to

pip install mylib

into the virtual environment where you'd start the

serve

process if you end up wanting to go the docker route, you'd just have to • create a work pool (I'd recommend push pool as mentioned earlier), but it could be

docker

too if you wanted to run docker on your VM • swap

.serve

for

.deploy

on your script ◦ pass in the reference to your

Dockerfile

to

.deploy

which does some

RUN pip install mylib

on top of the prefect base image

Ohh I didn't know we can pass in reference a Dockerfile to .deploy

Push pool is only with AWS, Google and Azure right ?

yeah gotcha, how much exactly it costs depends on how much runtime you use but if you've already got a VM then I'd just install docker there and start a docker worker there like i mentioned above w systemd

Yes Im gonna do that but I dont find the docs to add Makefile inside deploy... Im so sry for disturbing with that

no worries! what do you mean by Makefile here?

I think I miss understood something about how we make deployment. Because I have my script and at the end I have the deploy :

...python code before
@task
def set_last_update_date(current_date: datetime) -> None:
    supabase.table('recomend_tmdb_changes').insert([{
        'date': current_date.strftime("%Y-%m-%d"),
        'success': True,
    }]).execute()
    
@flow(name="tmdb_update_person", log_prints=True)
def tmdb_update_person():
    current_date = datetime.now()
    last_update_date = get_last_update_date()
    print(f"current_date: {current_date}")
    print(f"last_update_date: {last_update_date}")
    # tmdb_check_changes_person_ids(current_date,last_update_date)
    tmdb_check_export_person_ids(current_date,last_update_date)
    # set_last_update_date(current_date)

if __name__ == "__main__":
    flow.from_source(
    	source="<https://github.com/lxup/recomend-prefect.git>",
    	entrypoint="tmdb_update_person.py:tmdb_update_person",
    ).deploy(
        name="tmdb_update_person",
        work_pool_name="VPS_Worker_1",
        cron="0 6 * * *"
    )

so take this example

test.py

I can either install

pip install EXTRA_PIP_PACKAGES

at runtime or go ahead and pass a

DeploymentImage

like this so I can: • run this

python test.py

on my local • this will create the deployment on prefect cloud • create flow runs from my deployment that will be picked up by a worker (k8s work pool for me here, docker work pool for you on your VM)

Damn I think its working

good to hear! i think there are some native prefect features you could leverage before jumping straight to your own multithreading for example, I think you could rework some of your code to leverage task's

map

/

submit

interface, which allows you to leverage multithreading without implementing it yourself for example, this will run in about 5 seconds, as each task is executed in its own worker thread

My VPS with only 1vCpu and 2RAM dont use all power

i am not super familiar with running Prefect on VPS infra to be clear, the video just briefly covers task's

.submit

/ and

.map

which I think would be directly useful for optimizing your code in general I'd just recommend refactoring your code so that heavy IO / network bound stuff is in tasks that you can map (run concurrently) feel free to #C04DZJC94DC about VPS stuff, he'll probably know more than me 🙂

Well because right now my code for each person, make 2 request for getting details, and also 2 requests for uploading data in my database so if I can do multiple person in once like 5 directly it could be awesome. Im gonna check that. Btw Prefect is sick !

ahah wait Marvin is an ai agent ? 😭

yep! if you're curious here's the code

Im stupid, my actual script make for each items 3 supabase request, so during the script my supabase instance is completly overloaded. So I should store locally all items details, and after make one single Supabase request for update everything

There is a way to run multiple deployment in chain ? For exemple I have a main flow in tmdb_update.py, and I wanna call multiple flow (in tmdb_update_person.py for exemple)

from prefect import flow

from prefect.blocks.system import Secret

@flow(name="tmdb_update", log_prints=True)

def tmdb_update():

print("Starting tmdb_update")

tmdb_update_person()

# tmdb_update_person()

if __name__ == "__main__":

flow.from_source(

source="<https://github.com/lxup/recomend-prefect.git>",

entrypoint="flows/tmdb_update.py:tmdb_update",

).deploy(

name="tmdb_update_person",

work_pool_name="VPS_Worker_1",

cron="0 6 * * *",

job_variables={

"env": {

"EXTRA_PIP_PACKAGES": "supabase tqdm more_itertools",

}

}

)

is it possible ?

when you say “multiple deployments in a chain” do you mean you’ve already created some deployments and you want to call them from your tmdb flow and have them show up there? would you want to run them in the same process as your (parent) tmdb flow, or as a separate process? what you’re asking for is definitely possible, just trying to hone in on exactly what you mean

Sry @Nate for the late, well Im remaking all my script because it was completly unsecure (during the night all my database was broken because of the script ahah thx the backup). So now Im making all my script into one. Because I need to make run before other (like updating language, countries before adding new movie because movie table need relation with language et and countries table)

hi Loup, its a busy day for me, i can look more at this later. in general you cannot call tasks in tasks (arbitrary hard rule for now). subflows are infinitely nestable, so its just a question of what modularization is best for your use case (make tasks things you want to map or retry) can come back to this later

No problem Im gonna try myself I dont wanna force u to help me ahah🤣 And yeah the problem is if map can be only done with task I have to create flow...

there's also this pattern which is a way around that

Hello, my script crash because Ive reached the rate limite : Response: {'detail': 'Orchestration API rate limit reached. To raise your rate limit please contact help@prefect.io'} How can I fix that ? I dont understand when api are made because I dont really need them. I only wanna launchy my script automatically

api calls are made whenever you're actively running a flow for example, setting the states of flow or task runs, creating the flow / task runs themselves etc

Yeah I guess, I have this main flow in my script :

@flow(name="tmdb_update", log_prints=True)
def tmdb_update():
	print("Starting TMDB update")
	tmdb_update_language()
	tmdb_update_country()
	tmdb_update_keyword()
	tmdb_update_collection()
	tmdb_update_company()
	tmdb_update_person()

And each subflow contain task. For exemple there is the

tmdb_update_person()

: As u can see the

get_tmdb_person_details

is a task because I call it with map inside the flow for making parallele task...

how often are you running this flow?

My script should run every day at 8 am UTC

hmm i’d be surprised if running this 1 or a couple times a day caused you to hit rate limits

But maybe because my the task is running around 5 000 time just for one flow ?

yeah 5000 task runs within a short time would cause that on free tier. can you show your code specifically for that part?

I ca give u all my script :

please the section that’s involving the mapping / submission of 5k tasks would be more helpful :)

Yep but in my script I have 7 flows they all gonna make around 2-3k task... But for the one flow for exemple :

You can slow down task submission with a global concurrency limit to avoid triggering rate limits: https://docs.prefect.io/latest/guides/global-concurrency-limits/#use-cases

Ohh I’m gonna check that but we can’t just disable api call for a task ? I don’t really know how work the task system to make parallel task but in my case I don’t need log from task, only the parallel system

If the theory is that logging api calls are putting you over the rate limit and you would be all good if you disabled them, you can see what happens if you run the following command to disable logs.

prefect config set PREFECT_LOGGING_TO_API_ENABLED=False

if that doesnt work I do think. my colleague’s suggestion above is the most elegant.

Yeah I probably should use the first method. The example show this :

@flow
def my_flow():
    for _ in range(100):
        rate_limit("slow-my-flow", occupy=1)
        my_task.submit(1)

But in my case im using

.map()

so its the same usage ? There is a part of my code :

// INSIDE FUNCTION (FLOW)
...
for chunk in chunks:
				current_collections_to_update = []
				collection_details_futures = get_tmdb_collection_details.map(chunk)

The

get_tmdb_collection_details()

is a task so should I put the rate_limit function just before the map ?

catjamdid you get it work by using

rate_limit

?

I have to admit I dont know if is thank to the

rate_limit

or this command 🤣 :

prefect config set PREFECT_LOGGING_TO_API_ENABLED=False

I have the

rate_limit

function before all my

.map()

function

great to hear you’ve got it working!

Yeah ! Last question (after I stop asking u dummy question ahah), how can I add access_token for a private github repository ? The docs say this :

from prefect import flow
from prefect.runner.storage import GitRepository
from prefect.blocks.system import Secret

my_flow = flow.from_source(
    source=GitRepository(
        url="<https://github.com/org/repo.git>",
        access_token=Secret.load("github-access-token").get(),
    ),
    entrypoint="flows.py:my_flow",
)

my_flow()

But I have an error :

GitRepository.__init__() got an unexpected keyword argument 'access_token'

hmm I think

GitRepository

accepts

credentials

- not

access_token

With access_token : https://docs.prefect.io/latest/api-ref/prefect/flows/?h=from_source#prefect.flows.Flow.from_source

ahh that is a mistake in our docs, apologies.

GitCredentials

accepts

credentials

not

access_token

as a kwarg. you should be able to do

from prefect import flow
from prefect.runner.storage import GitRepository
from prefect.blocks.system import Secret

my_flow = flow.from_source(
    source=GitRepository(
        url="<https://github.com/org/repo.git>",
        credentials=Secret.load("github-access-token"), # note im passing the loaded block itself
    ),
    entrypoint="flows.py:my_flow",
)

my_flow()

i will open a PR to fix that docs example

Im getting an error with :

source=GitRepository(
        	url="<https://github.com/lxup/recomend-prefect.git>",
			credentials=Secret.load("github-access-token")
		),

`ValueError: Please provide a

token

or

password

in your Credentials block to clone a repo.`

great! yeah we can definitely make the docs better here 😅 thanks for sharing what worked for you!

Nahh thx you for helping me all this time ! 🙏

no problem! 🤘