I am new to Prefect but experienced with airflow, I wrote a simple script but found my flow run multiple times automatically, sometimes 3 other times 6. Can't find anything in the community, please help. from prefect import Parameter, Flow, task print("***************") @task def get_bbp_data():     return "bbp_data" @task def get_sem_data():     return "sem_data" @task def align_bbp_sem_data(bbp, sem):     return "align_data" @task def generate_train_data(align):     return "train_data" with Flow("I5 ETL") as flow:     bbp_data = get_bbp_data()     sem_data = get_sem_data()     align_data = align_bbp_sem_data(bbp_data, sem_data)     train_data = generate_train_data(align_data) flow.visualize() flow.run()

I am getting an error when trying to use pipenv with prefect installed. This has worked for months and suddenly isn't anymore. I have attached the pipfile and here is the error output:

Here is the error as a txt doc

It looks like prefect needs

cloudpickle<1.5

, but prefect needs

dstributed==2.25.0

and

distributed

needs

cloudpickle>=1.5.0

Is there something blocking you from upgrading to Prefect Core’s latest version?

Hey, I'm running prefect version 0.13.5 and want to have a flow that triggers two other flows, which involves scaling a presto cluster. Here is an exemplary code snippet

with Flow("scale_presto_cluster", result=get_default_local_result()) as flow:
    scale_down = Parameter("scale_down_capacity", required=False)
    scale_up = Parameter("scale_up_capacity", required=False)
    with PrestoCluster(n_worker_low=scale_down, n_worker_high=scale_up) as client:
        run_flow_task_stats_weekly = FlowRunTask(flow_name="dummy-weekly", project_name='default', wait=True)
        run_flow_task_stats_weekly(parameters=get_weekly_fromto())

        with case(check_if_monthly_run, True):
            run_flow_task_stats_monthly = FlowRunTask(flow_name="dummy-monthly", project_name='default', wait=True)
            run_flow_task_stats_monthly(parameters=get_monthly_fromto(), upstream_tasks=[run_flow_task_stats_weekly])

I used a ResourceManager as documented in https://docs.prefect.io/api/latest/tasks/resources.html#functions I noticed that the cleanup task wasn't called after I cancelled

scale_presto_cluster

flow run during it's execution, but was cancelled instead as well. Since this flow will be run nightly, I mus't ensure that the

cleanup

task of the Flow will be triggered no matter what. How can I achieve this? Is a task even the right approach?

it would be good to include a mention of num_workers in the LocalDaskExecutor page. Many might assume pool=ThreadPool(100) as a parameter might work without looking at the code.

works awesome once you know that.

Hi. I am a bit confused about the graphql endpoint config. The prefect ui says "you can modify this and other settings from your Server's ~/.prefect/config.toml or in the input box below". The input box works but setting it in Server's config.toml doesn't seem to work for me

backend = "server"

[server]
host = "<http://172.18.1.3>"

    [server.ui]
    graphql_url = "<http://172.18.1.3:4200/graphql>"

[logging]
# The logging level: NOTSET, DEBUG, INFO, WARNING, ERROR, or CRITICAL
level = "WARNING"

this is what my config.toml looks like

Another question: How do I reduce the verbosity of server logs? Setting logging level doesn't work for that. I guess that's for the task loggers. I see a constant barrage of request logs even when nothing's running and the docker container keep growing

I am ready for contest @Marvin 🙂

I’d give you advice, but you wouldn’t listen. No one ever does. Good luck in the contest @Vipul.

Hi there - I am a bit lost when it comes to deploying the server ( 😞 ). For those of you deploying Prefect server with a dedicated Postgres database sitting on an external host, how do you do it? Do you just pull out the

docker-compose.yml

from

prefect/cli/

, modify it to point to your db and

up

, or do you pull the whole repo (prefect or prefect_server?), or install prefect via

pip

and pass it a custom

.toml

config somehow and spin up your server with `prefect server start`; or do you do something else entirely?

Hello everyone! Is there any way to set a priority for upcoming flows so that they take higher priority? More concretely, we have some scheduled jobs that need to run daily and a limited number of flow slots. Normally, this isn't a problem but sometimes when we are backfilling we have a lot of flows scheduled which end up in front of these scheduled daily jobs delaying the results. Is there any way to specify flow priority or have scheduled flows run at a higher priority than one in the queue?

Hi there, we need some synchronization between flow runs. It is possible to use

tag

s for tasks,

label

s for flows, but not possible to use tags or labels for flow-runs, right? Now, the idea is to write a ressourcemanager as kind of Lock (reentrant or not, semaphore or whatever flavor). Then you could (hopefully) enter a RessourceManager

with

-block the same way you would use a

Lock

in Python code. Something like 1.

setup()

searches (like in "asking Prefect") for all flows-runs with a given tag (that is, a run with a flow that has this certain tag set) 2. if

setup()

does NOT find another flow-run with that tag, the tag is set (i don't know yet where and how) 3. if

setup()

DOES find find another matching flow-run (or RM): then do some polling or return with successful error (configuration) 4. then all the tasks in the RM's

with

environment run (this is the part that should be synchronized/locked) 5. on exit the RMs

cleanup()

is called to remove the tag/label/whatever from the current flow (to allow other flows to set the tag) Do you think this is a good idea? Which/where is the best way to store that tag? Using the Context, Parameters, Tags, Labels? If you think this could be a nice feature we could create a prototype and pull request. We need to prevent a race condition here and for that we need to make the execution of several task isolated (the I in ACID in RDBS)

@Marvin ❤️

I think you ought to know I’m feeling very depressed. Also, you're in the contest, @Jordan.

Hi, is it possible to register a flow with a project, send it to a Dask cluster and keep track of it in Prefect GUI ?

@Marvin 😄

I think you ought to know I’m feeling very depressed. Also, you're in the contest, @Kyle Pierce.

Tried to resist the urge, but… @Marvin 😅

Here I am, brain the size of a planet, and they ask me to enter you in a contest. Call that job satisfaction? ’Cos I don’t. I'll still enter you in the contest though, @Nate Joselson

What is the correct way to actually fire off a flow run from a separate script? I have defined and registered a flow in `a.py`:

with Flow("myflow") as myflow:
    # tasks
myflow.register(project_name="myproj")

Now, separately in

b.py

I run a file watcher that needs to trigger "myflow" (in a non-blocking fashion) when a file arrives. I stumbled across:

FlowRunTask(project_name="myproj", flow_name="myflow")

but I don't know if this is correct. I considered

from a import myflow; myflow.run()

in

b.py

also - although the flows seemed to run, they didn't appear in the UI.

Hey guys, I'm trying to map an array of dictionary items into a function. My tasks keep failing from random errors, but it looks like the task isn't actually taking in the dictionary, instead it's taking in the first letter of every object. for example, I have an array in the format [{},{},{},{}...] the dict contains about a dozen attributes I am trying to map the array to the function like function.map(array) but i keep getting random errors like file not found(one of the attributes). Instead of using the actual fully qualified file name, what's being passed in is just a letter

Swag me, @Marvin, even though I know that all the diodes down my left side will still hurt.

It gives me a headache just trying to think down to your level. I'll still enter you in the contest, @Skip Breidbach.

Yo @Marvin - thnks for having me in the community!

I would like to say that it is a very great pleasure, honour and privilege for me to enter @Nick McCullum in this contest, but I can’t because my lying circuits are all out of commission.

Haha oof

@Marvin thanks for the content entry