I am running an agent through supervisorin a remote server/ I am specifying the path for prefect to run.

command=/home/ubuntu/project/env/bin/prefect agent start --work-queue "my queue"

This works fine but then when i run flows in the UI the worker is using the global python runtime and no the one in the virtual environment, please help

If I wanted to parameterize a flow, how would I do that? I'd like to pass an array of strings (email addresses). The flow below works, but I'm not sure it is the proper way to do things. Please ignore the class name, just trying to test.

from pydantic import BaseModel


class CustomList(BaseModel):
    data: list[str]


@flow(name = "Fidelity Allocations")
def FidelityAllocationsFlow(toEmail: CustomList):
   #code....

if __name__ == "__main__":
    emailAddress = CustomList(data=['<mailto:cgunderson@spiderrockadvisors.com|cgunderson@spiderrockadvisors.com>'])
    FidelityAllocationsFlow(toEmail=emailAddress)

Hey all - having some Twitch API issues so we'll be live on YouTube only: https://www.youtube.com/c/PrefectIO

Hi All, Running prefect server which is to be accessed via load balancer what URLs should be used in config. Should it be load balancer url or host and port. Same question for agent. Thanks

Hey, on prefect 2 cloud I randomly encounter "internal server error"s when the agent wants to communicate with the server (exception see in thread). I have a flow which spawns up to 50 Tasks (the tasks are all done within seconds). I run the flow currently every 5 minutes. Am I running in some sort of api limitation due to too many tasks? Is there something to prevent this internal server error from happening? This error happens approx. every 15th time, all other times, the flow runs just fine.

File "/usr/local/lib/python3.9/site-packages/prefect/client.py", line 226, in raise_for_status
    raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Server error '500 Internal Server Error' for url '<https://api.prefect.cloud/api/accounts/bd169b15-9cf0-41df-9e46-2233ca3fcfba/workspaces/f507fe51-4c9f-400d-8861-ccfaf33b13e4/task_runs/|https://api.prefect.cloud/api/accounts/bd169b15-9cf0-41df-9e46-2233ca3fcfba/workspaces/f507fe51-4c9f-400d-8861-ccfaf33b13e4/task_runs/>'
Response: {'exception_message': 'Internal Server Error'}

Hello, with several long running flows I get an error like:

prefect.infrastructure.process - Process 'meticulous-manatee' exited with status code: -9

Any guidance?

hi there, in prefect 1.0 we could dynamically assign the task names with input params, something like:

@task(name="extract(ctx.id)")
def extract(ctx):

These would then be displayed in the UI as

extract(4)

for example if our flow was running for ctx.id=4. I haven't been able to find anything like this for 2.0, am I just missing something?

What is the relationship between an agent running on kubernetes and a kubernetesJob defined in infrastructure... • Does the kubernetes agent start the kubernetes job? ◦ if so, do I need a kubeconfig in the infrastructure block when the agent is already on the cluster? • If I run the job in an infrastructure process block from a kubernetes agent, will it run on the kubernetes agent?

Hey team. Can someone help with releasing the latest version of prefect-gcp? https://github.com/PrefectHQ/prefect-gcp/pull/61 The current version has a bug with credentials that I think will be solved with the commit from yesterday by @Andrew Huang

I’m getting the following error:

AttributeError: 'Credentials' object has no attribute 'project'. Did you mean: 'project_id'?

It looks like your PR will fix this @Andrew Huang.

Has anyone here tried running pyspark inside a prefect task? I get

RuntimeError: Java gateway process exited before sending its port number

when I run pyspark methods inside a task.

Hey guys, I’m running a flow with a DaskTaskRunner that spawns a task that takes around 20seconds and apparently it’s running sequentially?! My machine has 16 threads and I see in the UI each task only being run after another one finishes. Any inputs?

23:04:54.057 | INFO    | prefect.engine - Created flow run 'optimal-snail' for flow 'test1'
23:04:54.057 | INFO    | prefect.task_runner.dask - Creating a new Dask cluster with `distributed.deploy.local.LocalCluster`
23:04:55.980 | INFO    | prefect.task_runner.dask - The Dask dashboard is available at <http://127.0.0.1:8787/status>
23:05:00.371 | INFO    | Flow run 'optimal-snail' - Created task run 'Execute values of the query-8165e3c8-0' for task 'Execute values of the query'
23:05:00.372 | INFO    | Flow run 'optimal-snail' - Executing 'Execute values of the query-8165e3c8-0' immediately...
23:05:24.612 | INFO    | Task run 'Execute values of the query-8165e3c8-0' - Finished in state Completed()
23:05:26.370 | INFO    | Flow run 'optimal-snail' - Created task run 'Execute values of the query-8165e3c8-1' for task 'Execute values of the query'
23:05:26.370 | INFO    | Flow run 'optimal-snail' - Executing 'Execute values of the query-8165e3c8-1' immediately...
23:05:42.533 | INFO    | Task run 'Execute values of the query-8165e3c8-1' - Finished in state Completed()
23:05:44.295 | INFO    | Flow run 'optimal-snail' - Created task run 'Execute values of the query-8165e3c8-2' for task 'Execute values of the query'
23:05:44.296 | INFO    | Flow run 'optimal-snail' - Executing 'Execute values of the query-8165e3c8-2' immediately...
23:06:09.538 | INFO    | Task run 'Execute values of the query-8165e3c8-2' - Finished in state Completed()

As you see in these logs it’s not really spawning more tasks. I was under the impression that the call to a function with the

@task

decorator would be non-blocking if made within a flow

@flow

which uses DaskTaskRunner

Hi everyone! Just getting started with orchestrating DBT (CLI) flows using Prefect. Having a weird error where the "trigger_dbt_cli_command" is telling me I'm specifying an invalid project directory because I'm missing a dbt_project.yml file (I'm not). Any idea what's going on? Sample code below:

from prefect import flow
from prefect_dbt.cli.commands import trigger_dbt_cli_command

@flow
def run_dbt() -> str:
    result = trigger_dbt_cli_command("dbt run", project_dir='C:\\Users\\36350admin\\arrow_dbt-1')
    return result

run_dbt()

It even tells me it's running this dbt command, and when I run that by itself in the same working directory, it works as expected!

22:29:08.084 | INFO    | Task run 'trigger_dbt_cli_command-321ca940-0' - Running dbt command: dbt run --profiles-dir C:\Users\36350admin\.dbt --project-dir C:\Users\36350admin\arrow_dbt-1

Hi, it seems like DockerContainer block does not support ECR image yet? My agent is failing to authenticate with ECR. I'm basically doing something like this https://github.com/anna-geller/prefect-docker-deployment/blob/main/blocks/dockerhub_block.py but with ecr

Hi! I have a flow which only calls two subflows (and nothing else). The subflows show up as completed but the main flow is stuck at running... I've had related problems that seemed to be because of memory issues, but my subflows don't return anything so I don't see how they can complete without the main flow completing

Hello. Is it possible to stream task runner initialisation logs to the Cloud UI logs? I am using DaskTaskRunner and user can't see logs about cluster creation and dashboard url.

Hey there, I have a question about the "right way" to run prefect agents. I have the following setup: local environment (where I quickly test flows and tasks), github repo with CI which automatically redeploys new code, remote machine (which clones code from github, runs prefect deployment build and then starts prefect agent). And finally I am using Prefect Cloud for all remaining things. I am wondering that would happen If I redeploy my codebase (adding new flow and deployment for instance) and at the same time I would have some flows running. Would these flows be able to resume at the "right point"? And in general, what is the best practice for redeploying the codebase with prefect if I use only one machine? Thanks a lot! P.S. I used Airflow heavily before Prefect and now I am migrating some jobs to Prefect.

the prefect 2.x documentation regarding infrastructure such as for cloudrun and kubernetes job is way too scarce.

Hello, I have a problem when migrating to prefect cloud 2.0. i have a virtual machine and after running pip install -U prefect, pip doesn't work anymore, any lib I want to install doesn't work and the following message is displayed: AttributeError: module 'lib' has no attribute 'X509_V_FLAG_CB_ISSUER_CHECK' I tried to uninstall prefect and install again with the updated version and still the same error.

Hello! I use prefect 1 for flows that write data in a NFS, with Kubernetes and Docker runs, and docker storage for each flow. It seems that all flows run as root in their container as all files are owned by root when generated by the flow. I need to change the ownership of files generated, is it possible to change the user that flows execute as ? Thanks 🙂

Hi! does anyone know if it is possible to have multiple users and workspaces using a self-hosted prefect orion server? and if so, how could i install this server? thaaanks

When I host the ui locally (Orion) I don't get the opportunity to create new blocks - is this integration only supported in cloud version?

I have a question and an idea with regards to caching: 1. How can I forceful clean up cache without an expiry time? 2. If I have a task dependency: A->B->C and both A and B are cached, will there be any result transfered from A to B? In theory I would be able to only get cached returned values from B and run C. 3. It would be nice to have configurable cache locations by means of env var for each flow run. Currently when using input hash and with multiple users run the same task, prefect will try to fetch the cache from the first users, which raises permission issues depending. As a workaround I can get add the user to the hash function, but that feels complicated. Any thoughts?

Hi All, Am trying to use the prefect-email collection to send a notification via office 365 and getting the below error. Would anyone know how to fix this issue?

File "/usr/local/lib/python3.9/site-packages/prefect_email/credentials.py", line 138, in get_server
    server = SMTP_SSL(smtp_server, smtp_port, context=context)
  File "/usr/lib64/python3.9/smtplib.py", line 1050, in __init__
    SMTP.__init__(self, host, port, local_hostname, timeout,
  File "/usr/lib64/python3.9/smtplib.py", line 255, in __init__
    (code, msg) = self.connect(host, port)
  File "/usr/lib64/python3.9/smtplib.py", line 341, in connect
    self.sock = self._get_socket(host, port, self.timeout)
  File "/usr/lib64/python3.9/smtplib.py", line 1057, in _get_socket
    new_socket = self.context.wrap_socket(new_socket,
  File "/usr/lib64/python3.9/ssl.py", line 501, in wrap_socket
    return self.sslsocket_class._create(
  File "/usr/lib64/python3.9/ssl.py", line 1041, in _create
    self.do_handshake()
  File "/usr/lib64/python3.9/ssl.py", line 1310, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLError: [SSL: WRONG_VERSION_NUMBER] wrong version number (_ssl.c:1129)

This message was deleted.

I'm seeing some odd behavior in the Cloud UI that is not happening locally. I have

2.4.0

installed locally. For example, if I create the deployments

Foo(customer=1)

,

Foo(customer=2)

and

Foo(customer=3

all under the flow name

Foo

, I can go to my local UI, clicked on "Flows" and then "Foo", and I see those deployments. But in the Prefect Cloud 2.0 UI, if I go to "Flows" and click on "Foo", no deployments appear! They do appear if I click on deployments. Is this the expected behavior of the UI? (i.e. is there some criteria for a flow deployment to appear on the flow page?)

Question regarding upgrading Prefect 1.0 -> 2.0 Container Execution: All of my Prefect 1.0 flows run Docker containers, they aren't flows in Docker containers so in the course of upgrading to Prefect 2.0, the Docker Infrastructure block seems counterproductive. Do I need to rewrite all of the Docker Client tasks from Prefect 1.0 to maintain their functionality?

Hi all! I'm currently developing a flow with async and sync tasks. Tasks A and B run successfully but the execution freezes and I even can't see the log for task c, the last log is

Task run 'b' - Completed()

. If I force task c to be async runs without a problem. Is that a current bug? Using Prefect 2.3.1

@task
async def a():
  # do async

@task
async def b():
  # do async

@task
def c():
  # do sync

@flow
async def main():
  a = await a()
  b = await b(wait_for=[a])
  c = c(wait_for=[b])

asyncio.run(main())

Thanks!

is it possible to connect and deploy to prefect through azure when hosted locally like it is with prefect cloud?