Just to share, Disabling JIT on the PostgreSQL running the Prefect DB, significantly reduces peak load on the database (see screenshot). I ran the Prefect DB with pg_stat_statements to identify where our timeout issues came from. A query starting with

WITH RECURSIVE typeinfo_tree ...

was by far the heaviest. Several places online this was reported to be related to JIT (e.g. https://github.com/MagicStack/asyncpg/issues/530) so I attempted to disable JIT (

-c jit=false

when starting the PostgreSQL docker container) ... and then all the peak loads disappeared from our server. (Prefect 2.7.12)

Hi! Lately, we have had issues with a “flow file not found” error when running flows. This happens even when we simply rebuild an image that has worked previously. I see some other mentions of similar issues here. We use Azure storage for our flows. Has there been any recent changes in Prefect that might be relevant here? Our images are built from the

python:3.9

image, and we have tried to keep them lightweight by just pip-installing prefect. This has worked nicely enough with Kubernetes, although it has never worked when using Azure ACI. I added a post on discourse, about building more sturdy but still lightweight images. I imagine this is our underlying issue, and some recipes might be valuable for others as well. https://discourse.prefect.io/t/minimal-prefect-dockerfile-from-a-base-image/2502

Hi folks, I'm new to prefect. I've just written my SQLAlchemy models, hoping to query them from a

@task

, but I wrote the models using SQLAlchemy v2, and it looks like Prefect depends on v1.4. Is it not possible to use the latest SQLAlchemy version for my models with Prefect?

Hi! How can I get in touch with someone to switch to invoice as a payment method? I tried reaching out from my account but I did not get a reply.

Hi, is it possible to specify what files to upload to s3 when applying a deployment? I can see the files from the folder where the deployment is done are uploaded.. can I upload files from parent folders?

Hey community, Is it possible to get the

prefect logger

inside a

task runner

? I'm using the

RayTaskRunner

and I've defined a custom class using the following

class AnyscaleTaskRunner(RayTaskRunner):
def __init__(self, **init_kwargs: Any):
    some code

with Prefect flows and I have some information that is only available within the

AnyscaleTaskRunner

class. Is there any way that I can access the prefect context or the prefect logger in the

AnyscaleTaskRunner

class that I created?

Hi! I am using DockerContainer as infrastructure. Does it by default support concurrent task runs (via task.submit() ) or do I need to change some config? I got it working in process but not with DockerContainer, the tasks just runs in sequence.

After updating to 2.8 we found that each sub flow in a parent flow no longer scales to separate Kubernetes pods. Any idea on why this is happening?

Hi Everyone, Is there a way to DELETE flow code for a deployment stored in an s3 storage block? (

prefect deployment apply ... --upload

) If a new file in our codebase doesn't exist in s3, that file will get uploaded. ✅ If a file in our codebase gets deleted, I would like it to also delete in s3. ❓ Is this possible?

Hey all, I've spent the last two days trying to learn the ropes of Prefect, and I've made a lot of progress, but I have a feeling I may be using prefect-sqlalchemy fundamentally the wrong way. Would someone mind taking a quick glance at my workflow and telling me if I'm on the right track, or way off? https://gist.github.com/timwis/50fe4e90e052475e01d357e9b6142374#file-sync-py-L37 (In particular, I have a feeling the way I'm using Sessions is wrong, because the docs for prefect-sqlalchemy don't mention how to use it with ORM, so I took a bit of a leap)

I am a little confused about the package dependencies when using S3 as a storage block and then have an agent running somewhere else, read the code from there. Does the server with the agent have to have all the dependecies pre installed or can i somehow pull them from the bucket and have them installed? thank you

Dear Prefect Community, Please share an example to use my_completion_task_hook or my_completion_flow_hook using on_completion tag. I have the basic example from 2.8.3 release. What I am looking for is, is there a way to pass additional parameters along with task, task_run, state? For example, if I need to pass a specific error message in a task_failure_hook. Please help. If you need more details to help me, please let me know.

I'm seeing odd behavior with the script below. It's producing many flow runs(I'm expecting one) and none of the tasks are being run(they're omitted from the code block example). fwiw, the script is being run inside a lambda container. This lambda is being triggered from an event produced by a different lambda. Any ideas?

def handler(event: dict, context: LambdaContext):
    """This function is the entry point for the Lambda function"""

    <http://log.info|log.info>("starting lambda function for artist ranking")

    <http://log.info|log.info>(f"event: {event},context: {context}")

    start = time.time()

    deployment = Deployment.build_from_flow(
    flow=artist_ranker_etl,
    name="artist-ranker-deployment"    
    )
    deployment.apply()
        

    run_deployment(name="artist_ranker_staging/artist-ranker-deployment",tags=["staging","ranker"])
    
    
    end = time.time()
    
    print(f"Time to execute: {end - start}")

Hi there, I'm trying to figure out how I would leverage the caching functionality to ensure that any dependent tasks are re-run after a parent is re-run. 🧵

I'm also trying to figure out how I can defer execution of potentially expensive tasks in a flow unless requested, or to only execute a given task and its parents. 🧵

Hello guys, I’m trying to figure out why I have this problem:

socket.gaierror: [Errno -3] Temporary failure in name resolution
<https://api.prefect.cloud/api/accounts/my_account/workspaces/my_workspace/task_runs/0b3215a1-c402-4116-8320-439ca875a59a/set_state> failed

I use last version of prefect and my runner is RayTaskRunner. Infrastructure block is k8s. Any idea? Stacktrace in threads! ``````

Hi everyone, I thought to ask about issue regarding prefect 2 giving ec2:DescribeVpcs permission to ECStask infrastructure block. I would like to avoid giving this extra permission and I found Anna’s solution here https://discourse.prefect.io/t/why-the-ec2-describevpcs-aws-iam-policy-is-needed-on-ecstask-even-when-vpc-id-is-explicitly-provided/2062 but unfortunately when I try to deploy and run flow I am still getting error that I need this permission. Only thing I see different how I build ECStask block is that I am using task_definition_arn instead of registering new task definition, but I guess that's not the reason why I am getting error. Maybe someone have some ideas why this error is still appearing ?:)

Would importing inside/outside a flow would have an effect on performance?

Hey there 🙂 Is there any way in Prefect 1.0 to prioritize task for the execution? I have a flow with 1k tasks and some dependencies. Most of time hundreds of task are ready for execution, but i use just 20 workers (DaskExecutor). These tasks can be grouped and i would like to finish one group before continue with the next group. This could be easily done, if I would be able to set a priority for each task. I guess if this is not possible I have to build these groups with tasks for just modelling this groups with dependencies. Like: • 1: Start group 1 • 2-5: Task a,b,c (waits for 1) • 6: End group 1 (waits for 2-5) • 7:Start group 2 (waits for 6) • 8-10: Task x,y,z (waits for 7) • 11 End group 2 (waits 8-10) But this is a little bit ugly. Thanks for any help! Sadly we can not migrate to orion so soon... 😞

Hi i am trying to set up prefect using docker but not able to make connection with the /api and getting this message

Can't connect to Orion API at <http://0.0.0.0:4200/api>. 
Check that it's accessible from your machine.

This is my docker-compose which i am using to create the required prefect services.

version: "3.9"
services:

  database:
    image: postgres:15.2-alpine
    container_name: database
    restart: always
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=orion
    expose:
      - 5432
    volumes: 
      - db:/var/lib/postgresql/data
    networks:
      - backend
     
  minio:
    image: minio/minio:latest
    container_name: minio
    entrypoint: ["minio", "server", "--address", "0.0.0.0:9000", "--console-address", "0.0.0.0:9001", "/data"]
    volumes:
      - "minio:/data"
    ports:
      - 9000:9000
      - 9001:9001
    networks:
      - backend
     
  orion:
    image: prefecthq/prefect:2.8.2-python3.11
    container_name: orion
    restart: always
    volumes:
      - prefect:/root/.prefect
    entrypoint: ["prefect", "orion", "start"]
    working_dir: "/root/flows"
    volumes:
      - "./flows:/root/flows"
      - "./flows/tmp:/tmp"
      - "./flows/tmp_folder:/tmp_folder"
    working_dir: "/root/flows"
    environment:
      - PREFECT_UI_URL=<http://127.0.0.1:4200/api>
      - PREFECT_ORION_API_HOST=0.0.0.0
      - PREFECT_ORION_DATABASE_CONNECTION_URL=<postgresql+asyncpg://postgres:postgres@database:5432/orion>
    ports:
      - 4200:4200
    depends_on:
      - database
    networks:
      - backend
 
  agent:
    image: prefecthq/prefect:2.8.2-python3.11
    container_name: agent
    restart: always
    entrypoint: ["prefect", "agent", "start", "-q", "queue_1", "-q", "queue_2", "-q", "queue_3"]
    volumes:
      - "./flows:/root/flows"
      - "./flows/tmp:/tmp"
      - "./flows/tmp_folder:/tmp_folder"
    environment:
      - PREFECT_API_URL=<http://127.0.0.1:4200/api>
    networks:
      - backend
    
volumes:
  prefect:
  db:
  minio:
networks:
  backend:
    driver: "bridge"

Can anyone please check where i am doing wrong ... Thanks

Hello. In prefect2, I have different behavior locally vs a cloud flow run: I'm defining a few flows with the same functionality like this (I'm using

Flow

instead of

@flow

):

def create_flow(flow_name, param):
    def general_func(param):
        do_something(param)

    return Flow(general_func, name=flow_name)

my_flow_1 = create_flow(flow_name="my_flow_1", param=ex_1)
my_flow_2 = create_flow(flow_name="my_flow_2", param=ex_2)

Now this works fine locally. But when I deploy to the Prefect cloud, I get this error when I run the flow (deployment is fine):

raise MissingFlowError(prefect.exceptions.MissingFlowError: Flow function with name 'general_func' not found in {my directory}.

Why is it expecting a flow called

general_func

? How can I fix this? p.s. I don't want my general_func to be a subflow. Thanks

built a docker image for infrastructure block to run flows. When run, I'm getting the following. Anyone know how to get past it? Seems like I'm missing something fundamental

File "/home/ubuntu/apps/dataflows/venv/lib/python3.11/site-packages/prefect/infrastructure/docker.py", line 95, in _get_docker_client
    raise RuntimeError(f"Could not connect to Docker.") from exc
RuntimeError: Could not connect to Docker.

Hi everyone, I’ve been running into an

NotImplementedError

when using the prefect_shell collection. It was working previously but for some reason when I tried doing it again it’s not working. I’ve already used pip install -U prefect-shell and double checked that its installed correctly in the venv Lib But when running a simple test I get the following “Not Implemented” error: The source code can be found below:

from prefect import flow
from prefect_shell import shell_run_command

@flow
def example_shell_run_command_flow():
return shell_run_command(command="ls .", return_all=True)

if name == "main":
print("test")
example_shell_run_command_flow()

error message from the cloud UI is attached.

Error_msg▾

1569×931 120 KB▾

Any help would be appreciated. Thanks!

I am attempting to migrate prefect1 flows to prefect2. My prefect1 flow uses LocalDaskExecutor as my executor, Docker as my storage, and KubernetesRun as my run_config. It seems like prefect2 no longer supports docker as a storage class. What is the best way to store flows that span multiple files using prefect2?

Hello everyone. I was following the Contributing Guide, but when I run

pytest tests

I get:

ImportError while loading conftest '/Users/daniel.mesejo/PycharmProjects/prefect/tests/conftest.py'.
tests/conftest.py:67: in <module>
    from prefect.testing.cli import *
src/prefect/testing/cli.py:7: in <module>
    from prefect.cli import app
src/prefect/cli/__init__.py:9: in <module>
    import prefect.cli.cloud
src/prefect/cli/cloud.py:58: in <module>
    login_api = FastAPI(on_startup=[set_login_api_ready_event])
../../opt/anaconda3/envs/prefect/lib/python3.8/site-packages/fastapi/applications.py:124: in __init__
    self.router: routing.APIRouter = routing.APIRouter(
../../opt/anaconda3/envs/prefect/lib/python3.8/site-packages/fastapi/routing.py:502: in __init__
    super().__init__(
../../opt/anaconda3/envs/prefect/lib/python3.8/site-packages/starlette/routing.py:592: in __init__
    warnings.warn(
E   DeprecationWarning: The on_startup and on_shutdown parameters are deprecated, and they will be removed on version 1.0. Use the lifespan parameter instead. See more about it on <https://www.starlette.io/lifespan/>.

This is because the latest version (0.94.0) of FastAPI uses the version 0.26.0.post1 of starlette, that it turn introduces the

DeprecationWarning

. The last update for FastAPI was done yesterday March 10th (see here). In the short term downgrading to 0.93.0 (

pip install fastapi==0.93.0

) solves the issue, but perhaps for the future is worth creating a ticket to start using the lifespan parameter. What would be the best course of action here? In any case I’ll be happy to help 🙂.

Is it possible to persistently annotate/attach metadata to a successful deployment run in Prefect with metadata or should I use storage for that? We use Prefect to asynchronously launch a short-lived workflow (<1min) from a Node.js backend. This allows us to free the Node.js backend from being blocked. The output of this workflow is a simple JSON with an array of key-value pairs. We would like to be able to return the result of this JSON in the simplest possible way. What’s the best practice here? AFAICT there are 3 options: 1. Persist the result into a database from the workflow 2. Persist the result into a storage 3. Annotate the deployment run / attach metadata to the deployment run using Prefect SDK If option (3) exists it would be great, less responsibilities on the application side of things to setup the database, persist the data and then apply garbage collection to stale data (e.g. when deployments get deleted, we need to do a cascaded delete where the JSON body also needs to be wiped out). Your thoughts? Thanks!

Hi all! Does anyone know the best practise for catching custom exceptions within prefect tasks? I have an exception which isnt a code failure, its just data not being available, and I’d like to send a slack warning when this happens but not report flow failure. I hoped that this would work:

try:
        data = task.submit(start, end, logger=logger)
    except DataWarning as e:
        logger.warning(e)
        send_incoming_webhook_message(
            slack_webhook,
            f"Warning raised for {task_name} for {start} to {end} due to {e}",
        )

But no dice, doesnt play well with the future object. Im reworking all of this to not use exceptions and just pass dictionaries around, but thought knowing the best practise might be useful

Hi everyone 👋 Is there any plan to support infrastructure configuration at runtime with flow parameters and not only at deployment time with deployment specs or at initialization time with task runner configuration? Currently, we have to define the flow pod k8s resources when we deploy it and dask task runner specs before the flow is executed, but having a way to configure both with flow parameters would be convenient!

Hi everyone, I was trying to use Kubernetes Job infrastructure block for flow so that I can add the s3fs and other dependent packages, but getting this error in prefect-agent, when trying to run flow. can someone help me with this?

kubernetes.client.exceptions.ApiException: (400)
Reason: Bad Request
HTTP response headers: HTTPHeaderDict({'Audit-Id': 'a78a7ce9-1b45-439f-824f-b5470e93dea2', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'Date': 'Sat, 11 Mar 2023 18:22:02 GMT', 'Content-Length': '179'})
HTTP response body: b'{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"strconv.ParseInt: parsing \\"59.911360159516335\\": invalid syntax","reason":"BadRequest","code":400}\n'

Hey all, I am trying to migrate from prefect version 1 to prefect version 2. In my prefect 1 script, i used to create dynamic tasks using the names which are read from a text file. since, prefect 2 does not allow a task to call another task, i tried to call the function which present in another task and tried to create dynamic task using the name present in the text file. The names in the text file is printing but could not create a task using it. Please, help me in this. Thanks