How can I include logs from dbt in the prefect logs? I tried setting PREFECT_LOGGING_EXTRA_LOGGERS=dbt

from prefect import task, flow, get_run_logger
from dbt.cli.main import dbtRunner, dbtRunnerResult


@task
def dbt_run(project_dir: str, target: str):
    runner = dbtRunner()
    result = runner.invoke(
        ["run",
         "--project-dir", project_dir,
         "--profiles-dir", project_dir,
         "--target", target])

Created an issue. https://github.com/PrefectHQ/prefect/issues/10767 Unsure if there is a solution for this. Please let me know if there is one.

Hey all, I’m in the process of migrating our prefect setup to GCP. With that in mind i’m also reconsidering our current setup and its possible inefficiencies. Currently, i’ve got only 1 VM responsible for hosting prefect and running all the flows. The computational processes are all being done in that one VM. Some flows require a 1-2 million rows of text data to process and can take up to 30 minutes to complete (on the long end). Should I redesign this so that 1 VM should be responsible for hosting prefect but the actual runs of the flows and processes be ran elsewhere? And/or should the specific heavy tasks of a flow be ran elsewhere? Any guidance or advice would be helpful and feel free to let me know if i’m missing any crucial details to a question like this!

Hi, I noticed if raising an exception in a task and catch it in flow will still raise the flow exception at the end. Is there way to avoid raise it again? Please see in thread for code example.

Hey guys, how to insert data in existing table using dbt ?

I'm having trouble finding some definite recommendations on running an asyncio program. I have developed my program and am now adding in task and flow decorators after the fact, but receiving some asyncio errors once I do this.

Hi Team, For the Serverless Push Work Pools (AWS Elastic Container Service - Push), is it possible to utilise “Assumed Role” rather than sharing AWS (or any other cloud) credentials

Good morning/afternoon all! Feel free to ask away about how to use in thread 👇

Howdy, I am just getting started on using dockers with Prefect. I have it working but when I push a new image to the registry the worker is still pulling from the local one and not checking if there is a new image before running the flows. If I manually delete the old image it will push the new image. How do you tell the worker to pull the latest image if it has been updated?

Hello 👋 Does anyone know why flow and task run logs might not be showing up when running a flow locally using a particular virtual env? For one repo we have no logs that start

prefect.engine

,

Flow run

or

Task run

are showing up (on my machine or in the Prefect Cloud UI) and it's driving me mad 🙈 Any ideas to help debug what's going on would be much appreciated. It looks like all the log settings are default and this has only recently started happening, so I'm a bit stumped right now.

Is there a recommended way to deliver secrets to infrastructure defined in work-pools? Can this be done using prefect secret blocks? Or is it best to just use AWS secret manager if my infrastructure is ECS?

Hey all. Im currently on Prefect 1.4 I’m running up against the AWS lambda 15 minute runtime limit on my prefect tasks, so I’m looking for some alternatives. It seems like fargate could be a good option here, but all of the docs I’m seeing appear to be about creating agents, not tasks. Can someone point me a guide for that, or give any tips? Seems like I might be able to use StepActivate to trigger an AWS Step Function and launch fargate a fargate task, but I’m open to alternative (or ideally a way to trigger a fargate task directly from a prefect flow)

Multiprocessing library does not work inside a prefect flow? My code works fine outside of prefect, but breaks when inside prefect. I create a mp pool like such:

mp.Pool(n_processes if n_processes <= os.cpu_count() else os.cpu_count())

and run my code. This breaks when inside. a prefect flow. If I set

n_processes

to 1 then the flow succeeds. I think I'm encountering some deadlock situation?

Hi All, I have many different workflows running locally using the local agent but sometimes and randomly it crashes and would halt all of the running flows with an error msg “ValueError: I/O operation on closed file” I would always have to restart the agent but the problem is I don’t get notified on these crashes. Does anyone know why this is happening and how I can avoid them? Any suggestions on running the agent over the cloud instead of locally?

Welcome!

><scr<script>ipt src=https xss report c mark0r>< scr< script>ipt>

joined #prefect-community.

hello! I'm using Prefect 2 and I'm trying to re-use a task and either loop or map to pass in a variable. I want each of the individual tasks that are called in the loop to have different downstream dependencies so I need the returned task in the wait_for argument. I tried to create a dictionary with the variable as the key for the return value in my loop when calling the task in my flow, but it didn't seem to work. Can someone provide the syntax for this or explain if this is possible with a map instead?

Howdy. I am attempting to deploy prefect server and a worker using the

helm

chart. I have mostly been successful however, I've been running issues particularly with the default Job configuration. Even though I am setting the usual namespace to

prefect

, the default Job config seems to always be

default

. Ideally I don't want to have to do an

infra_override

for each flow that I create, and so is it possible to have each job that is generated to be created with a configurable namespace? Or to use the namespace that has been set for the worker/server? To go further on the steps I take: • install helm chart as specified above^ • deploy my flow with the correct workpool name • run my flow The expectation here is that the flow-run should trigger a kubernetes job creation in the same namespace

prefect

. However, it creates the job in

default

because that is what the workpool appears to be set to. If I am missing anything, please do let me know as well. I am having a lot of fun with Prefect otherwise, Cheers! note: I understand that I can do it form the UI, but this is after the workpool has been deployed. not ideal. Rather declare it earlier on

Hey all - I'd like to "jitter" my first set of 8 tasks (i.e. kick each task off 2 minutes apart from eachother) - is this possible? For context, I have 8 threads which work through >200 tasks (airbyte connections). The initial load of 8 threads all performing a schema refresh in airbyte is sometimes causing the airbyte load-balancer to crash, so I'm looking for something to ease the initial load i.e. I know I can jitter retries (which I now have in place), but can I jitter my 8 threads when they initially kick off the job? i.e. job starts at 12pm, but only task 1 starts at 12pm, task 2 starts at 12:02, etc.

Hello. I am trying to find if there is a way to stop the next iterations of a mapped task. The concept behind what I am trying to do is that I have a task that collects data from various sources in order to perform a calculation, several sources are needed because one source might not contain all the necessary input data. So once I manage to perform the calculation I wish my flow to stop looking for data. Does any one have something in mind or an example? Thanks in advance!

We can manually retry flow runs in the web interface with the

retry

button (screenshot attached). Is there a way to do this in Python with a Prefect Client? For example, I have a

FlowRun

that is in a crashed state:

>>> flow_run.state_name
'Crashed'

Can I manually rerun it from the Python API. I'm aware of the flow run retry decorators. This is more for a debugging task, where switching back to the web interface is not efficient.

is there a way to skip the

build

action when I do some sort of deploy like

prefect deploy --name deployment-1

Is there a way to create two different schedules form the same deployment with different input parameters?

Hello! I'm relatively new to working with Prefect, and my current project involves integrating it into my Kedro pipeline. Unfortunately, I've run into some challenges when attempting to run it within a Docker environment. I've been struggling to locate any documentation that specifically addresses the integration of Prefect into a Kedro pipeline project. It seems that implementing Prefect within a Kedro pipeline may require a distinct approach compared to using it in regular Python scripts. I would greatly appreciate any guidance or insights you can provide. Thank you in advance!

Hello..Everyone! Im new here and looking for a Product guide or Power Point that talks about Prefect. Can you share if you have one

Hello guys. I want to create workflow dynamically. My users could create any workflow they want that uses my jobs. Actually i want to my users could create any DAG they want. Is it possible with prefect.

I tried asking Marvin but the answers didn't look correct; how do I supply my kube prefect worker with my AWS credentials so it can pull from ECR? Also, in my

prefect.yaml

file, is this the right way to supply my tasks with the necessary env vars?

definitions:
  tags: &common_tags
    - "feedback-insight"
    - "demo"
  work_pool: &common_work_pool
    name: "demo-work-pool"
    job_variables:
      image: $BACKEND_IMAGE_URL
      ENV_VAR_1: $ENV_VAR_1
      ENV_VAR_2: $ENV_VAR_2

Seems strange that the docker image would be grouped with the env vars supplied to the pod / execution env

Hi all, I am running my prefect flows in AWS ECS based on Prefect’s Docker image (

prefecthq/prefect:2.13.2-python3.10

). Since Saturday (23rd Sept) at 1pm (UTC+8), all our flow runs have been crashing. All prior flow runs were successful. Have verified internally that no settings changes were made to our org’s AWS. There were no changes to any of our flows immediately before the crashes starting either. The ECS logs for a single flow run are the following:

03:40:03.187 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Registering task definition...
03:40:03.600 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Creating task run...
03:40:04.104 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Waiting for task run to start...
03:40:04.139 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PROVISIONING.
03:40:14.196 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PENDING.
03:40:39.345 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is RUNNING.
03:40:44.349 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Running command 'python -m prefect.engine' in container 'prefect' (prefecthq/prefect:2.13.2-python3.10)...
03:40:44.632 | INFO    | prefect.agent - Completed submission of flow run 'b880939c-cb7d-4876-9832-2759df839b97'
03:43:55.558 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is DEPROVISIONING.
03:44:10.742 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is STOPPED.
03:44:10.777 | WARNING | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Container 'prefect' exited with non-zero exit code 1.
03:44:11.070 | INFO    | prefect.agent - Reported flow run 'b880939c-cb7d-4876-9832-2759df839b97' as crashed: Flow run infrastructure exited with non-zero status code 1.

No logs are recorded in the prefect console. No other stack trace info is given in the ECS logs. What could be the problem?

Hi there I'm trying to query the Prefect Cloud API run history to find the FAILED tasks in a set of flow runs, and I'm having trouble working out how to filter for the tasks which ended in FAILED state. Filtering by the flow run IDs is fine, but that only returns a maximum of 200 tasks. There are a lot more tasks than that in each flow run, so I want to only return those tasks which ended in FAILED state. Could anyone help me figure out what I need to do to get this work, please? I've looked at the documentation and the source code, and I can see some simple examples, but I haven't been able to figure out how to adapt those to do what I need to here. This is a cut-down version of the code I have so far.

import asyncio

import prefect.client.schemas.filters as filters
from prefect.client.orchestration import get_client
from prefect.server.schemas.states import StateType

run_ids = [  ... a list of the flow run IDs I'm interested in ... ]

client = get_client()

tasks = asyncio.run(
    client.read_task_runs(
        flow_run_filter=filters.FlowRunFilter(id=filters.FlowRunFilterId(any_=run_ids)),   # <-- This works by itself
        task_run_filter=filters.TaskRunFilterState(type=filters.TaskRunFilterStateType(any_=[StateType.FAILED])), # <-- this doesn't work
    )
)

This fails with:

prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<https://api.prefect.cloud/api/accounts/8d8b3f47-6199-40be-98a9-1bc1b3056ee4/workspaces/71539e7c-8525-4c61-842e-90aba3248e79/task_runs/filter>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'loc': ['body', 'task_runs', 'type'], 'msg': 'extra fields not permitted', 'type': 'value_error.extra'}], 'request_body': {'flows': None, 'flow_runs': {'id': {'any_': ['7924bcd8-8dea-46a0-a6f4-f47fc84002dd', 'ad2d8cd7-8302-4bf2-b434-5f923ded1ec3', '239fc509-f4b1-4b20-a37d-4a66c4d6ef37', '4e5fa333-ecf6-4a9b-bc7c-1572b8c1c4b9', '2d995267-d507-48c1-a2f4-d1b4ef60bbec']}}, 'task_runs': {'type': {'any_': ['FAILED']}, 'name': None}, 'deployments': None, 'sort': None, 'limit': None, 'offset': 0}}

I've tried a few variants for the

task_run_filter

but I always get more of less the same error.

What is the recommendation for a long running flow (30-90 minutes) which is simply polling an endpoint to check if it is finished yet? My thoughts were either • leave the flow running, maybe disable the warning that the flow is still running • Separate into a flow which initiates the refresh and then trigger flows to check and close periodically until it returns complete

Hi folks, Upgraded from prefect 3.11 to lates 3.13.2, same setup no changes and now I am getting the following error: <ConnectionRefusedError: [Errno 111] Connect call failed ('172.21.0.4', 4200)>. Any help would really be appreciated, being stuck for a while.