prefect-community #best-practices-coordination-plane

Hi! We have a way we are using prefect that required a little bit of a hack to get working, and I'm wondering if people here can tell me if there's a better way! The issue is: we have a repo that contains some commonly used prefect flows (ill call it 'flow repo'), and we want to import those flows into several different repos, and deploy the flows from those repos (ill call them end-user repos). Our method is to define Deployment objects in the end-user repos and

.apply()

them. (we use s3 storage, and run our flows via Kubernetes jobs, in images which have the flow repo installed as a python package. This works fine except that when we build the Deployments in end-user repos, we run into the problem that the files containing the imported flows are not under the same working directory that we are building the Deployments from. The flow repo is installed as a python package and thus is under

/opt/env/.../site-packages/...

. This causes the code that sets the Deployment's entrypoint to fail with an error like

<path_to_flow_file> is not in the subpath of <cwd> OR one path is relative and the other is absolute.

. We are working around this by importing the flows and modifying their

.___module___

property to 'trick' prefect into thinking that the flows come from the end-user repo. Is there something I'm missing here that would allow us to do this in a less hacky way, or is this pattern just not something prefect supports? Thanks!

Richard Alexander

02/08/2023, 4:05 PM

I have a question about logging with async flows in Prefect 2. In short, if we start a flow like this:

Copy code

from prefect import flow, get_run_logger

@flow()
async def my_flow():
    logger = get_run_logger()
    # Start async process and pass logger

Will the logger also be async? As a bit of background, we have an async process that spawns multiple other async processes and will create a good deal of logging that we need to catch with the prefect logger. We want to make sure that the logging won't cause any blocking issues. Is passing a logger in this way a safe way to go about it for async code?

Jon

02/10/2023, 5:02 PM

questions about migrating from v1 to v2 on a sizable project: 1. is it easier to open a branch for the migration, or to open a new repo completely? 2. how much of a 1:1 is the infrastructure? i see that agents behave differently, and that we can set up infra via the UI. should we expect a total overhaul of what we have today? (we are using ECS)

Oluremi Akinwale

02/13/2023, 10:02 AM

Hi everyone, please how can I run prefect 2.0 Agent within python code that contain the flow?

Alireza

02/14/2023, 9:49 AM

Hi everyone, It seems

Great Expectation

is no longer supported in v 2.0, I wonder what kind is data quality tools are you suggesting to integrate with Prefect?

✅ 1

Stefan

02/17/2023, 9:41 AM

Copy code

def main(date: DateModel = None) -> None:

How can I use the current date for flow_run_name? Unless doing data for another day, I don't call the flow with any arguments.

Serina

02/17/2023, 4:46 PM

set the channel description: Want to understand the best way to use Prefect? Looking for examples of code patterns are people using? Look no further! Prefect Recipes: https://github.com/PrefectHQ/prefect-recipes

blob attention gif 3

🎉 3

Kelvin DeCosta

02/21/2023, 12:05 PM

Hey everyone! I have a question regarding

async

execution of tasks via

.map

. In the documentation for the

.map

function, it mentions that if the task that is being mapped is

async

, then the

.map

call must be awaited. However, I keep getting an error with the type checker stating that there are no overloads of

.map

that can be awaited. What should could be done to fix this?

✅ 1

Stefan

02/23/2023, 8:19 PM

Hi! How can I trigger another flow run from an existing deployment from a flow? Say I want to generate a flow run from Flow B as soon as Flow A finishes, but they are different and independent flows and deployments. Can I do it in Python, or an API call?

✅ 1

Alvaro Durán Tovar

02/27/2023, 10:45 AM

is there any date to shutdown prefect v1?

✅ 1

Austin Weisgrau

02/27/2023, 9:34 PM

What's a best practice look like for setting up hundreds of concurrent REST API calls using prefect tasks? We want to use some kind of concurrency limit to avoid overwhelming the API. However, the concurrency limits in prefect make tasks wait 30 seconds before checking for an available slot. This results in a HUGE slowdown, assuming each API call takes significantly less than 30 seconds a pop (less than 1 second each is normal).

Conrad Dobberstein

02/28/2023, 10:38 AM

Hello Everyone, I have a question regarding the behaviour of flow call parameters. In the example below, the flow

my_flow

takes an iterable of strings as the first parameter. When calling the flow with an iterator, its first element is missing inside the flow run. This also applies when calling the flow with a list and enabling

validate_parameters

. Is this a bug, or am I missing something obvious here? I'm on the latest version of Prefect, i.e. Prefect 2.8.3.

Copy code

from collections.abc import Iterable

import prefect


@prefect.flow
def my_flow(texts: Iterable[str]) -> None:
    print(f"{type(texts)=}")
    print(f"{list(texts)=}")


if __name__ == "__main__":
    # type(texts)=<class 'list_iterator'>
    # list(texts)=['World', '!']
    my_flow.with_options(validate_parameters=True)(["Hello", "World", "!"])

    # type(texts)=<class 'list_iterator'>
    # list(texts)=['World', '!']
    my_flow.with_options(validate_parameters=False)(iter(["Hello", "World", "!"]))

    # type(texts)=<class 'list'>
    # list(texts)=['Hello', 'World', '!']
    my_flow.with_options(validate_parameters=False)(["Hello", "World", "!"])

Jafar A

03/01/2023, 8:01 PM

I'm trying to deploy and run a job in docker and getting error about boto library that I included in docker block but no use in the code yet. what's happening?

Copy code

prefect deployment build src/main.py:ingest_everflow -n test-run-container  -q test
-sb s3/s3-everflow/ef_test_code -ib docker-container/de-container -o prefect-docker-deployment --apply

return self.sign(operation_name, request)

File "/usr/local/lib/python3.10/site-packages/botocore/signers.py", line 189, in sign

auth.add_auth(request)

File "/usr/local/lib/python3.10/site-packages/botocore/auth.py", line 418, in add_auth

raise NoCredentialsError()

botocore.exceptions.NoCredentialsError: Unable to locate credentials

✅ 1

Jafar A

03/02/2023, 8:33 PM

Hello, I've had good success deploying with prefect in subprocess, docker locally and also with ECR. Now when it comes to k8s, docker desktop k8s is running (green with k8s_POD_prefect-k8s-... running) but prefect says Pod never started! I followed a post by @Jeff Hale and this is my build:

prefect deployment build flows/main.py:ingest_everflow \

-n prefect-k8s-example -q k8s \

-sb s3/s3-everflow/ef_testing/docker-k8s-image \

-ib kubernetes-job/prefect-k8s-example \

--apply

Copy code

14:40:49.825 | INFO    | prefect.agent - Submitting flow run '99f81791-abda-488f-802e-d949535cb351'
/Users/jafaramin_1/env/lib/python3.10/site-packages/prefect/agent.py:216: UserWarning: Block document has schema checksum sha256:1a553852c1bcc9cd95645917bdbaaae9e28cbcc1270ff5b43b45bbd87564f5cc which does not match the schema checksum for class 'KubernetesJob'. This indicates the schema has changed and this block may not load.
  infrastructure_block = Block._from_block_document(infra_document)
14:40:50.538 | INFO    | prefect.agent - Completed submission of flow run '99f81791-abda-488f-802e-d949535cb351'
14:40:50.546 | DEBUG   | prefect.infrastructure.kubernetes-job - Job 'prefect-k8s-example-mwmgm': Starting watch for pod start...
14:40:50.555 | INFO    | prefect.infrastructure.kubernetes-job - Job 'prefect-k8s-example-mwmgm': Pod has status 'Pending'.
14:40:54.827 | DEBUG   | prefect.agent - Checking for flow runs...
14:40:59.925 | DEBUG   | prefect.agent - Checking for flow runs...
14:42:50.560 | ERROR   | prefect.infrastructure.kubernetes-job - Job 'prefect-k8s-example-mwmgm': Pod never started.

Kelvin DeCosta

03/06/2023, 8:41 AM

Hey everyone! I've started to run into this issue again: https://prefect-community.slack.com/archives/C03D12VV4NN/p1671793942895089 Setting

cache_result_in_memory=False

, seemed to help at first, but now, after about 32k tasks, the flow run crashes. The crash seems to because of this error:

Copy code

OSError: [Errno 28] No space left on device

Any help is appreciated!

Richard Alexander

03/07/2023, 3:15 PM

Quick question about

agents

work_pools

and

storage_blocks

. I have several agents (each polling a different work pool) on the same server trying to poll jobs from the same

s3

storage block, but only one of them is working. Can multiple agents/work pools be connected to the same storage block?

Daniel McNeil

03/08/2023, 11:12 PM

Question: is there a way to stop a local agent but waiting until any running flows are completed?

Jafar A

03/14/2023, 8:49 PM

Hello, Deployed and built using ECS agents and now I'm getting below error. ECS block got access to aws credentials, taskdefinition, roleARN and ..etc. any guess what this is about?

raise ProfileNotFound(profile=profile_name)

botocore.exceptions.ProfileNotFound: The config profile (prefect) could not be found

Frederik Vanhevel

03/14/2023, 9:33 PM

I'm trying to figure out how to use prefect to make around 25000 api calls but only execute 10 at a time. I know similar questions have been asked before but I can't figure out the right way how to do it so any help would be greatly appreciated! What i've tried so far: • The suggestion that i found here on slack using semaphore, this only works locally • Using custom code to limit conucrrency using asyncio • Previous options but with async definitions: https://docs.prefect.io/tutorials/execution/#asynchronous-execution Other questions I have: • Should i make 25000 tasks? • Should i have one task that makes all the calls? Appreciate the help!

❓ 1

davzucky

03/14/2023, 11:29 PM

Did you check the task concurrency limit or work queue? https://docs.prefect.io/concepts/tasks/#task-run-concurrency-limits https://docs.prefect.io/concepts/flows/

davzucky

03/14/2023, 11:29 PM

These concepts already exist in perfect to solve that

Kelvin DeCosta

03/15/2023, 12:12 PM

I've been having memory usage build-up issues with long-running tasks that call (via

.map

) many tasks (10k-100k). I'm using the

SequentialTaskRunner

and have noticed that the task runner keeps a

_results

dict of ALL task results. It even stores the results of tasks that are no longer "needed". What harm do I cause by running something like

get_run_context().task_runner._results = {}

, when I know that the previous task results are not needed?

Nelson Griffiths

03/15/2023, 5:24 PM

We are trying to find a good way to build up a dependency graph from deployments so that deployments all trigger the downstream deployments that depend on them. Are there any best practices around this? I realize I could do this with automations, but there are limits to how many automations a workspace can have. On top of that, it would also be nice to have these dependencies specified as code, and to my knowledge automations can not be specified as code. I would also love for this to not be structured as subflows so that it is easy for things to be triggered in different places and there is no requirement of importing a flow into a master flow. It would also be preferred that someone could make their flow depend on an upstream flow without going into the upstream flow and adding a line at the end of the flow like

run_deployment(...)

. Plus this way would limit a flow to depend on only 1 upstream flow from finishing. I have a few ideas on how we could potentially do this, but I was wondering if anyone here has set up something like this before? It seems like something prefect could be built to handle?

Jafar A

03/15/2023, 9:33 PM

Hello, Anyone seen this

crashed

in prefect UI happening ? I'm running ECS agent and UI error says :

Flow run infrastructure exited with non-zero status code 1.

17:26:06.079 | DEBUG   | prefect.agent - Checking for cancelled flow runs...

17:26:07.931 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'electric-guan': Status is RUNNING.

17:26:09.893 | DEBUG   | prefect.agent - Checking for scheduled flow runs...

17:26:12.938 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'electric-guan': Running command 'python -m prefect.engine' in container 'prefect' (<http://880954911558.dkr.ecr.us-west-2.amazonaws.com/prefect-docker-creditfurnishing:latest|880954911558.dkr.ecr.us-west-2.amazonaws.com/prefect-docker-creditfurnishing:latest>)...

17:26:12.971 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'electric-guan': Streaming output from container 'prefect'...

17:26:13.023 | INFO    | prefect.agent - Completed submission of flow run '57ccf19d-e50c-469e-9697-5a5b46011ba9'

17:26:13.059 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'electric-guan': Status is DEPROVISIONING.

exec /usr/bin/tini: exec format error

17:26:16.249 | DEBUG   | prefect.agent - Checking for cancelled flow runs...

17:26:21.936 | DEBUG   | prefect.agent - Checking for scheduled flow runs...

17:26:24.004 | DEBUG   | prefect.agent - Checking for cancelled flow runs...

17:26:30.948 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'electric-guan': Status is STOPPED.

17:26:31.550 | WARNING | prefect.infrastructure.ecs-task - ECSTask 'electric-guan': Container 'prefect' exited with non-zero exit code 1.

17:26:31.708 | DEBUG   | prefect.agent - Checking for cancelled flow runs...

17:26:31.726 | INFO | prefect.agent - Reported flow run '57ccf19d-e50c-469e-9697-5a5b46011ba9' as crashed: Flow run infrastructure exited with non-zero status code 1.

17:26:32.807 | DEBUG   | prefect.agent - Checking for scheduled flow runs...

17:26:41.023 | DEBUG   | prefect.agent - Checking for cancelled flow runs...

17:26:42.106 | DEBUG   | prefect.agent - Checking for scheduled flow runs...

Bebeto Nyamwamu

03/16/2023, 11:23 AM

when was this setting PREFECT_LOGGING_MARKUP introduced? Which version?

Nelson Griffiths

03/22/2023, 4:33 PM

I am digging into the automations API trying to create custom triggers and have run into a few questions: • For a custom trigger, can you match to a deployment? All the examples I see show a match on

prefect.flow-run.*

. But could I also match on something like

prefect.deployment.my-deployment

? And is there a list of the potential values for

prefect.resource.id

that can be matched on somewhere? • From Prefect's view is a schedule a specific implementation of a trigger. Like the event is the clock hits 7 am and then flows get started? Or are the treated differently in the backend? • Is there any plan to tie triggers(automations) to deployments? I know they are a general tool and need to be used for things other than deployments (such as notifications). But it seems to me that for event driven systems also allowing them to be defined with deployments as an alternative to a schedule would be a really good thing? Just curious if this is in the plans or if they will always exist separately. Nothing make or break here, just would be convenient to create a deployment with a trigger set up already. • Is Prefect orchestration the right tool for very event driven systems? Or is Prefect's aim less about that and I should be using something else to watch for events and kick off Prefect deployments via the API? I know there are ways to do this currently in Prefect. And it sounds like there will be some improvements to that soon. I am mostly curious if this is a big focus of Prefect moving forward or if its more of a side use case?

YSF

03/24/2023, 4:44 AM

Hey all, rookie here. What is the best practice for using blocks with varied teams? I have a very small team but an ultra rigid set of 'need to know' based access controls for things like docker images, code repositories, buckets/filesystems. I love the idea of reusable blocks but I'm apprehensive whether I can even use them in a workspace, it sort of feels like have a github repo with multiple .env files with keys that anyone can pull and use. For something like a storage block that gives you access to an s3 bucket doesn't it basically circumvent all other access controls for all members of the workspace? Does anyone have a pattern they're implementing that either allows for fine grained access control, or inherits existing auth mechanisms?

👀 2

Charlie Henry

03/24/2023, 4:37 PM

Building off ☝️, I'm curious what people are using to organize blocks to make them less messy? We're using github storage so we usually have to make a new storage block for every feature branch we want to test before merging to production, so things get crufty over time.

John

03/30/2023, 4:56 PM

Is it a good practice to run a flow that never ends? e.g. a web server

✅ 1

Max Jackson

04/04/2023, 6:54 PM

Does Prefect 2 have something analogous to

wait_for_flow_run

from Prefect 1? I'd like to have a flow of flows which stops in a Failed state if one of its subflows fails, but I'm not sure how best to implement this in Prefect 2.

✅ 1