prefect-community #prefect-community

Yufei Li

07/06/2023, 2:53 PM

Hi all, is there any tutorial about setting up opsgenie_webhook_block.notify? To be more specific, we'll need the integration to send the workflow logs to Opsgenie.

Lee Mendelowitz

07/06/2023, 5:27 PM

Is there a best practice for how to retry flow runs that crash? Our specific use case is for retrying flows that run as ECSTask’s on Fargate Spot.

✅ 1

datamongus

07/06/2023, 8:53 PM

Is there a roadmap for Go support

Daniel Lomartra

07/06/2023, 9:47 PM

Is it possible to create conditional/dependent parameter type validation appear in the prefect UI? For example, if I had two enumerators and managed a mapping between them, would it be possible to set up something where a users selection in one dropdown effects the values available in the second one?

Sandip Viradiya

07/07/2023, 12:51 AM

I'm setting up Prefect in ECS. Work queue shows healthy. I'm using Bitbucket as storage. I've added EXTRA_PIP_PACKAGES as

prefect-aws prefect-bitbucket

I am getting below error when I try to run any flow. Any ideas? Thanks in advance 🙂

Copy code

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 395, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.10/site-packages/prefect/client/utilities.py", line 51, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/deployments/deployments.py", line 193, in load_flow_from_flow_run
    storage_block = Block._from_block_document(storage_document)
  File "/usr/local/lib/python3.10/site-packages/prefect/blocks/core.py", line 618, in _from_block_document
    else cls.get_block_class_from_schema(block_document.block_schema)
  File "/usr/local/lib/python3.10/site-packages/prefect/blocks/core.py", line 672, in get_block_class_from_schema
    return cls.get_block_class_from_key(block_schema_to_key(schema))
  File "/usr/local/lib/python3.10/site-packages/prefect/blocks/core.py", line 683, in get_block_class_from_key
    return lookup_type(cls, key)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/dispatch.py", line 185, in lookup_type
    raise KeyError(
KeyError: "No class found for dispatch key 'bitbucket-repository' in registry for type 'Block'."

gojaeheon

07/07/2023, 6:37 AM

Hello~ I want to use a shared resource(variable) in a flow. There are many tasks in a flow and tasks want to share specific resource(variable like str).. What can i do for this?

Deceivious

07/07/2023, 8:33 AM

Hi, Can someone help me override Settings context? Usecase: I have 3 profiles: 1. local_db [Connects to self hosted postgres database] 2. dev [Dev prefect api] 3. prod [Prod prefect api] In addition to that, I want a unit test profile which connects to a dockerized postgres. I do not want this to be a profile as the tests will be run on multiple machines - hence the need for a fresh Settings context. We cannot use the test harness as we need to run database queries. How do I initialize my own settings context without creating new profile?

Nico Neumann

07/07/2023, 8:39 AM

Is there a way when starting a flow through the CLI (

prefect deployment run

) to watch the logs in the terminal? In prefect 1 there was a watch flag:

prefect run --watch

but I couldn’t find anything in prefect 2

✅ 1

Tim-Oliver

07/07/2023, 2:02 PM

Hi, quick question: How does the

ConcurrentTaskRunner

decide how many workers are used? Is there a way to manipulate this?

Junjun Zhang

07/07/2023, 2:57 PM

hi, greetings, with

prefect-databricks

is it possible to launch jobs using an existing cluster? the documentation only showed how to create new cluster

Javier Ochoa

07/07/2023, 4:02 PM

Hi, When using ECS Task feature from Prefect 2, do I still need to define the infrastructure on my own at AWS? or is there something that Prefect 2 configuration does for me? I am not following how it fully works

✅ 1

Charles Leung

07/07/2023, 4:29 PM

hey everyone, my prefect logs arent showing up on the UI, when i look into the agent it keeps displaying this error. any reason why?

✅ 1

ale

07/07/2023, 4:35 PM

Hey folks 👋 I think I have a use case where the (Prefect v1) Schematic view would be really useful in Prefect 2! A bit of context: I have my dbt projects, and each one is defined in a specific Prefect flow. Each dbt project represent a specific subject area. There are interdependencies between these projects, meaning that they have to run in a pre-defined way. An “orchestrator” flow takes care of triggering each dbt flow run. In the orchestrator we pre-define the dependencies using (Prefect v1)

set_upstream_tasks

(in Prefect v2 I would use

wait_for

). The result is that, once the orchestrator flow is deployed, the Schematic view provides a very clear and useful DAG of all dbt flows. In Prefect v2 there’s no concept of a pre-defined DAG, hence it’s not possible to have this kind of view before running the flow. And if you made a mistake in setting dependencies, you will only discover it at runtime, which is not nice 😅 However, for an orchestrator flow, I think it would make a lot of sense to have a sort of Schematic view similarly to what is available in Prefect v1.

🙏 1

Rio McMahon

07/07/2023, 4:36 PM

Question on workspaces: what is the best practice for leveraging workspaces in prefect 2 for project/process/environmental delineation? I have two competing ideas: 1. Use workspaces to isolate dev/prod: We have a flow called

widget-flow

that is mirrored between a dev workspace and a prod workspace. 2. To isolate processes:

widget-flow

has its own workspace with prod and dev versions of the flow within that workspace. My intuition is that 1) is “better” in that it reduces proliferation of agents/work queues. This also begs the additional question: can a single agent monitor multiple work-queues from multiple work queues? Thanks for any insights.

✅ 1

Jacob Goldberg

07/07/2023, 4:49 PM

In prefect 1.0 you could use the

upstream_tasks

argument to manually block one task on others, is there an equivalent in prefect 2.x?

✅ 1

skrawczyk

07/07/2023, 5:52 PM

Hey all, I'm looking for a way to dynamically turn off the schedules of old flow deployments using the Python API. For example, lets say I have a flow called

hello_world

with a deployment called

1.0

that is scheduled to run everyday. I already have some code in place that can automatically create a new version of

hello_world

as a deployment named

1.1

that would also run everyday. I want to retroactively go back and turn off the schedule for

hello_world:1.0

so there aren't 2 versions running at the same time. I see that the Deployment class has a

load()

and

update()

function, which would take care of turning off the schedule if I knew which deployment name I needed to turn off. Does anyone know how I can list all deployments a flow has stored and select the most recent or the ones that have active schedules?

✅ 1

Ajeel Ahmed

07/07/2023, 9:12 PM

super quick question: if I have two tasks in a flow and the first one fails the second one should still run yes?

✅ 1

Ajeel Ahmed

07/07/2023, 9:45 PM

okay so for anyone else who had this question this is the answer: if they’re just normal

tasks

no, the second task fail will also fail but if you run these tasks via

task_runners

and say use the sequential one, it’ll be fine and the second one will run even if the first one failed

✅ 1

Ajeel Ahmed

07/07/2023, 9:45 PM

correct me if I’m wrong of course

Bryan Rodas

07/07/2023, 10:19 PM

Any one else having issues with the Prefect Cloud IU? Can't seem to create a custom run - submit button not working

👀 1

Tom Klein

07/09/2023, 12:40 PM

Hello, We have an issue with prefect-dask where, if our tasks rely on a “complex” object being returned (e.g. one that cannot be pickled) - we have a problem moving from the default

ConcurrentTaskRunner

to the

DaskTaskRunner

. Basically we encounter the issue mentioned here: https://discourse.prefect.io/t/picklingerror-error-when-using-a-dasktaskrunner/908 What’s the best way to work around this, other than changing the object being returned (assuming that’s not feasible) ? is there some way to override how serialization is done on it? or - why does it work with the regular task runner but not with dask?

Kyle

07/10/2023, 4:51 AM

the concurrency by tag feature works great but i also need to rate limit some API requests to third parties by tag as well. is there something built into prefect that helps with this or do i need to use my own python code to implement this?

Kyle

07/10/2023, 5:02 AM

^ i think the solution that would work best is using redis with “ratelimiter” in python. i dont see anything in prefect docs so i assume i need something custom.

Copy code

import redis 
from ratelimiter import RateLimiter

# Connect to your Redis server
redis_conn = redis.Redis(host='localhost', port=6379, db=0)

rate_limiter1 = RateLimiter(redis_conn, max_calls=500, period=10, name='ratelimit1')
rate_limiter2 = RateLimiter(redis_conn, max_calls=1000, period=60, name='ratelimit2')

with rate_limiter1:
    try:
        response = requests.get(BASEURL)
    except Exception as e:
        msg = f"[{ticker}] API request failed: {e}"
        raise Exception(msg)

Deceivious

07/10/2023, 8:30 AM

HI, Does prefect support task retry dependancy [Ill explain ]? Consider the following use case. 1. Task 1: Fetches data from API. Returned junk data 20% of the time. Categorizing response as junk data is a complex CPU intensive parsing task. 2. Task 2: Parses the response and categorizes into good / bad data. Raises Custom Exception "BadDataException". In case Task 2 raises the exception, I want to return Task1 and Task2. I still want the 2 "task" to be 2 "Prefect Task" as I use result storage for task 1 and different task concurrency tag limiter for the 2 tasks. Id say something like

ChainedTask

if that makes sense 😄

Farhood Etaati

07/10/2023, 8:49 AM

Hi. How can I pass a

job_variables

object to a deployment object in python? The docs only talk about

yaml

and

cli

configurations for

work-pool

and base job templating.

✅ 1

Ajeel Ahmed

07/10/2023, 1:09 PM

Copy code

Loading flow for deployment 'backup_precog_database_flow'...
11:41:18 AM

Starting 'ConcurrentTaskRunner'; submitted tasks will be run concurrently...
11:41:18 AM

Executing flow 'backup-precog-database-flow' for flow run 'dynamic-rooster'...
11:41:19 AM

Beginning execution...
11:41:19 AM

Created task run 'backup_to_drive-9749801d-0' for task 'backup_to_drive'
11:41:19 AM

Executing 'backup_to_drive-9749801d-0' immediately...

Ajeel Ahmed

07/10/2023, 1:09 PM

why is it saying

Starting 'ConcurrentTaskRunner'; submitted tasks will be run concurrently...

when I’m not running any task runner?

Ajeel Ahmed

07/10/2023, 1:10 PM

I’m just calling simple two tasks in a flow without using

task_runners

jpuris

07/10/2023, 1:17 PM

Heya! We're trying to run flows on kubernetes party k8s via a Worker. All seems to be well, though we are struggling to find where to change the default

back off limit

for the job pods? It is currently set to 6 and we do not want the pods to restart at all, if any failure has occurred.

Luke Dolan

07/10/2023, 3:08 PM

Hey thought I'd post here first before creating a bug report When using read_flow_runs, the returned flow runs do not have the state expected by the given filter Server version 2.10.13 Client version 2.10.20 The simple code snippet I have

Copy code

import asyncio
     from prefect.client.orchestration import PrefectClient
     from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType
     from prefect.client.schemas.objects import State, StateType
 
    client = PrefectClient(api=f'<http://xxx.xxx.com:4200/api>')
    state_filter = FlowRunFilterState(
               type=FlowRunFilterStateType(
                   any_=[StateType.RUNNING]
               )
           )
    flows_resp = asyncio.get_event_loop().run_until_complete(client.read_flow_runs(flow_run_filter=FlowRunFilter(type=state_filter), limit=1))
     <http://logger.info|logger.info>(f'got {len(flows_resp)} flow runs')
    for flow in flows_resp:
         <http://logger.info|logger.info>(f'flow state details {flow.state}')

this gives the result

Copy code

16:54:32.780 | Info   | httpx - HTTP Request: POST <http://xxx.xxx.com:4200/api/flow_runs/filter> "HTTP/1.1 200 OK"
16:54:32.780032000 [Info ] [httpx] HTTP Request: POST <http://xxx.xxx.com:4200/api/flow_runs/filter> "HTTP/1.1 200 OK"
16:54:32.792552000 [Info ] got 1 flow runs
16:54:32.792795000 [Info ] flow state details Completed()

In the above I would expect that the flow state is

RUNNING

rather than

COMPLETED