Hello! I am using

run_deployment

to run sub-flows on their own pod/node when the parent flow is deployed on my Kubernetes work pool (in EKS). However, I'm noticing when the cluster scales up to support the scheduling of these new pods/nodes, prefect thinks these sub-flows crashed (when eventually they start and complete successfully) causing the parent flow to move on when it should be blocked. I'm just looking for a solid way for sub-flows to be scheduled on their own pods/nodes (due to heavy resources required). Is there a preferred way to do this? I've scoured the docs/discourse/faq to no avail... Here's a torn-down version of a parent-flow i'm working on:

@flow(log_prints=True)
async def my_flow(config_path: str, local=False):
    my_things = [ . . . ]

    if local:
        # local invocation -> process things sequentially
        for thing in things:
            await my_subflow(config_path, thing.id)
    else:
        # deployed invocation -> each subflow gets its own pod
        await asyncio.gather(
            *[
                run_deployment(
                    name="my-deployed-subflow",
                    parameters={"config_path": config_path, "id": thing.id},
                )
                for thing in things
            ]
        )

Hey everyone, how do I change the default path so that when I run something like "Prefect deploy .." it wont look in the default C:\ drive for my python.exe file? I have a weird usecase where my venv is in another machine and we access it via UNC path to the python.exe

@Marvin I have this flow run from a deployment defined below failing with the following error:

06:41:31.085 | ERROR   | prefect.worker.kubernetes.kubernetesworker acc28481-964b-45e9-a84f-25cdaa0529a1 - Flow run 2d2e82e1-2e37-4bc1-b25d-7ffc51ee637e did not pass checks and will not be submitted for execution
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 855, in _submit_run
    await self._check_flow_run(flow_run)
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 841, in _check_flow_run
    raise ValueError(
ValueError: Flow run UUID('2d2e82e1-2e37-4bc1-b25d-7ffc51ee637e') was created from deployment 'hello_flow_deployment' which is configured with a storage block. Please use an agent to execute this flow run.

Can you point how the deployment should be defined to work with my k8s work pool, having the flow being executed as a Cloud Run job?

from prefect.deployments import Deployment
from prefect_gcp.cloud_storage import GcsBucket
from prefect_gcp.cloud_run import CloudRunJob
from prefect.client.schemas.schedules import CronSchedule
from dataflows.flows.hello_flow import hello_flow

storage = GcsBucket.load("prefect-storage")
infrastructure = CloudRunJob.load("cloud-run-default")


def deploy_hello_flow():
    deployment = Deployment.build_from_flow(
        flow=hello_flow,
        name="hello_flow_deployment",
        work_queue_name="default",
        storage=storage,
        path="hello_flow",
        tags=["staging"],
        infrastructure=infrastructure,
        schedule=CronSchedule(cron="0 12 1 * *", timezone="UTC"),
    )
    deployment.apply()

Hey, I am having trouble running a dbt prefect flow, and the documentation around this is very confusing, bordering on misleading and incomplete. Here is my flow

import os
from prefect import flow, task
from prefect_dbt import DbtCliProfile, DbtCoreOperation, SnowflakeTargetConfigs
from prefect_snowflake.credentials import SnowflakeCredentials
from prefect_snowflake.database import SnowflakeConnector
from prefect.context import FlowRunContext

@flow
def trigger_dbt_data_build(client: str, warehouse: str = "COMPUTE_WH"):
    flow_run_name = FlowRunContext.get().flow_run.dict().get("name")
    credentials = SnowflakeCredentials(
        user=os.environ.get("SNOWFLAKE_USER"),
        password=os.environ.get("SNOWFLAKE_PASSWORD"),
        account=os.environ.get("SNOWFLAKE_ACCOUNT"),
        role="ACCOUNTADMIN",
    )
    connector = SnowflakeConnector(
        # schema=f"{client_name.upper()}_STG",
        schema="dbt_zmunro",
        threads=8,
        database="RAW",
        warehouse=warehouse,
        credentials=credentials,
        query_tag=f"dbt-data-build-{client}-{flow_run_name}",
    )
    target_configs = SnowflakeTargetConfigs(
        connector=connector,
        extras={
            "retry_on_database_errors": True,
            "connect_retries": 0,
            "connect_timeout": 600,
            "retry_all": False,
            "reuse_connections": False,
        },
    )
    dbt_cli_profile = DbtCliProfile(
        name="prefect-snowflake-dev",
        target="dev",
        target_configs=target_configs,
    )
    return DbtCoreOperation(
        commands=[
            f'dbt build --select +somemodel --vars \'{{"client_schema":"{client.upper()}_STG"}}\''
        ],
        dbt_cli_profile=dbt_cli_profile,
        overwrite_profiles=True,
        project_dir="/workflows/dbt/",
    ).run()

I am getting an error because the

DbtCoreOperation

cant find my

dbt_cli_profile

correctly. I am trying to follow the documentation here: https://prefecthq.github.io/prefect-dbt/cli/credentials/#prefect_dbt.cli.credentials That documentation doesn't say how to use this

dbt_cli_profile

object with the

DbtCoreOperation

function though. And I am confused as to how blocks relate to this all as well. Should I not be creating these credential objects each flow run and instead just do it once and save it in a block? where does this block get saved if I am not using prefect cloud?

Hi, I have set up a cloud hook that is supposed to notify me via MS Teams when a run is successful. Unfortunately, I don't get any notification and the test call doesn't seem to work either. I also use the same webhook URL in the Airbyte installation, the message also arrives in MS Teams, so that can't be the problem. Can anyone help here? By the way, I use Prefect Server. Thanks in advance

@Marvin I have a prefect flow running 10 tasks in parallel on Ray via EC2. When I execute the following code by

python 3 file.py

, it works perfectly. I want to create a deployment out of this that I can run and trigger from the UI. I am happy to use docker for this and upload my docker file to a registry. I am on prefect 2.x and want to stay on it. How?

import time
from prefect import flow, task
from prefect_ray import RayTaskRunner

@task
def shout(number):
    time.sleep(0.5)
    print(f"#{number}")

@flow(task_runner=RayTaskRunner(
        address="<ray://1.2.2.3:10001>",
        init_kwargs={"runtime_env": {"pip3": ["prefect-ray"]}},
    ),
)
def count_to(highest_number):
    futures = [shout.submit(number) for number in range(highest_number)]
    for future in futures:
        future.wait()  # Ensure all tasks complete

if __name__ == "__main__":
    count_to(10)

@Marvin how to make a deployment with kubernetes infastructure with python

Hello, I'm on latest prefect 2.10.6, and I have a lot of tasks that hang indefinitely, I'm running the tasks in parallel inside a flow with a tag concurrency limit and a timeout of 60 seconds set for each task, but it still hangs indefinitely until I kill it, more details in the thread ⬇️

We have batch jobs written in other languages and packed into container and write our task mainly using

RunKubernetesJob

. But everytime I run the flow using

KunernetesAgent

, the agent always spin up new job, which in my case is useless and waste of resource. Since the flow is mainly about spin up new k8s job, I want

KubernetesAgent

run it directly. Please advise me the way to do that.

How do I run concurrent flows? I understand that tasks can run concurrently, but I have flows that are already compositions of tasks, and the flow needs to run concurrently for sets of different parameters.