I am trying to executed concurrent tasks of flow from multiple agents with same work queue, but only single agent is picking the tasks. Is there a way to run these tasks on multiple agent with same queue? Using prefect - 2.6.4 #prefect-community #prefect-contributors

Hi everyone ! Is it possible to save/update only part of a

JSON

block ? I did see nothing like that in documentation. To give some context, I got multiple concurrent tasks depending on a common block and actualizing only different part of it. As I don’t want to lose data and because of the concurrency, I want to avoid loading the

JSON

block in each tasks, saving it just after with my updates and risking to overwrite it with missing value. So my question here is can I save only part of the

JSON

block ? Or maybe there is another solution (like lock or something) to prevent block data loss when it comes to async parallel task ?

Hi. After upgrading agent to 2.7.1 getting the error provided in thread.

Hello everyone. I am experimenting a bit with the dask taskrunner, and I am running into some issues related to priority. I have the following test-flow:

from typing import List, Tuple

import dask
from prefect import flow, get_run_logger, task
from prefect.context import get_run_context
from prefect_dask import get_dask_client


def is_prime(number: int) -> Tuple[int, bool]:
    if number == 2 or number == 3:
        return number, True
    if number % 2 == 0 or number < 2:
        return number, False

    for i in range(3, int(number ** 0.5) + 1, 2):
        if number % i == 0:
            return number, False

    return number, True


@task
def get_primes_from_split(min_number, max_number) -> List[int]:

    if min_number % 2 == 0:
        min_number += 1

    with get_dask_client() as client:
        futures = [client.submit(is_prime, n) for n in range(min_number, max_number, 2)]

        maybe_primes = [future.result() for future in futures]

    return [value for value, flag in maybe_primes if flag]


@flow(name="example_prime_number_search")
def main(max_number: int = 1_000_000, split_size=10_000):
    log = get_run_logger()
    context = get_run_context()

    <http://log.info|log.info>("Task Runner: %s", context.task_runner.name)
    <http://log.info|log.info>("Searching for primes from up to %d", max_number)

    futures = [get_primes_from_split.submit(x, x + split_size) for x in range(0, max_number + 1, split_size)]
    primes = [value for future in futures for value in future.result()]

    if len(primes) > 10:
        <http://log.info|log.info>("Found %d primes: %s, ...", sorted(primes)[::-1][:10])
    else:
        <http://log.info|log.info>("Found %d primes: %s", sorted(primes)[::-1][:10]

When running this with the

DaskTaskrunner

, the task

get_primes_from_split

is scheduled first, then the dask-future

is_prime

. Since

get_primes_from_split

is scheduled first it gets higher priority, which causes the dask-execution to lock up, as it is waiting for the task to complete before executing anything else.

get_primes_from_split

naturally is waiting for

is_prime

to complete, which unfortunately will not execute at this point. Toying around with priorities, I managed to get

is_prime

to execute:

from typing import List, Tuple

import dask
from prefect import flow, get_run_logger, task
from prefect.context import get_run_context
from prefect_dask import get_dask_client


def is_prime(number: int) -> Tuple[int, bool]:
    if number == 2 or number == 3:
        return number, True
    if number % 2 == 0 or number < 2:
        return number, False

    for i in range(3, int(number ** 0.5) + 1, 2):
        if number % i == 0:
            return number, False

    return number, True


@task
def get_primes_from_split(min_number, max_number) -> List[int]:

    if min_number % 2 == 0:
        min_number += 1

    with get_dask_client() as client:
        futures = [client.submit(is_prime, n, priority=100) for n in range(min_number, max_number, 2)]

        maybe_primes = [future.result() for future in futures]

    return [value for value, flag in maybe_primes if flag]


@flow(name="example_prime_number_search")
def main(max_number: int = 1_000_000, split_size=10_000):
    log = get_run_logger()
    context = get_run_context()

    <http://log.info|log.info>("Task Runner: %s", context.task_runner.name)
    <http://log.info|log.info>("Searching for primes from up to %d", max_number)

    with dask.annotate(priority=0):
        futures = [get_primes_from_split.submit(x, x + split_size) for x in range(0, max_number + 1, split_size)]
        primes = [value for future in futures for value in future.result()]

    if len(primes) > 10:
        <http://log.info|log.info>("Found %d primes: %s, ...", sorted(primes)[::-1][:10])
    else:
        <http://log.info|log.info>("Found %d primes: %s", sorted(primes)[::-1][:10])

This causes dask to schedule a few instances of

get_primes_from_split

, which in turn schedules all its instances of

is_prime

.

is_prime

executes properly and starts returning its results, but it doesn't seem like

get_primes_from_split

picks up execution. I really don't understand what is going on here. Can anyone provide some insight into how do this kind of execution without reaching a deadlock like above?

Hi all, I’m looking how to migrate over to Prefect 2.0 and am designing my deployment pattern. Is there an equivalent to the extract_flow_from_file function in Prefect 2.0?

Within prefect 1.0 is it possible to have multiple slack webhook URLs for different slack channels? e.g. we want to delineate between data engineering and data science prefect alerts (different alerting channels for each). Looking at the docs: https://docs-v1.prefect.io/core/advanced_tutorials/slack-notifications.html#installation-instructions it seems like there is only one secret per account called

"SLACK_WEBHOOK_URL"

.

Question regarding resource management in Prefect 2.x when all tasks are async (i.e. “submitted”) and some tasks are optional, in 🧵

Hi all - Not sure why this error is occurring. I've have included s3fs, gcsfs, and adlfs in the Poetry toml file used to create the Docker image.

Hi all, how do you guys check if Block exists inside a Flow? I tried wrapping with

storage_name = f"{self.account.name}-session"
try:
    self.session = JSON.load(storage_name)
except ValueError as e:
    self.logger.error(e)
    self.session = JSON(value={})
    self.session.save(name=storage_name)
    <http://self.logger.info|self.logger.info>(f"created session storage ({storage_name=})")

but flow keeps failing on this exception.. How to gracefully handle it? ADDED: seems the problem was capital letters in block name! otherwise code works as intended. fixed by modifying type in model definition for account.name to constr(to_lower=True)

Could someone help me understand if a Prefect Flow supports my use case? I’m building something similar to Plaid. I’m planning on using a Prefect Flow to orchestrate my per-user data pipeline. Here’s a simplified version of what it looks like: 1. Extract the user’s bank transactions. This involves logging into their bank with their submitted credentials, scraping their bank transactions, and saving this raw data to bucket storage. 2. Transforming the bank transactions. This involves downloading the raw data from bucket storage, transforming the data into our standardized format, and saving this data to a postgres table. This Flow must be run on a per-user basis. There are two instances in which this Flow is ran: 1. When a user is first created 2. During a daily job to refresh all user’s data Thus, if I have 1M users, then I would need something that supports 1M concurrent Flow Runs. Is this achievable with Prefect?

I'm having trouble registering my flows with S3. I saw on the prefect discourse that there is an undocumented

upload_options

argument to the

S3

block, which is great because I need to specify

ServerSideEncryption

, however, I'm receiving

AccessDenied

botocore.exceptions.ClientError: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied

Hello All , I got weird problem with Prefect 1.0 flow run , I have other flows running fine with same server and same Agent(s) . I made a small change on this one and got new version , but still getting error .

17:06:30 INFO Agent1

Submitted for execution: PID: 3327309

17:06:31 ERROR execute flow-run

Failed to load and execute flow run: PermissionError(13, 'Permission denied')

Does anyone have any links to docs/code for setting up a Prefect Orion webserver for testing client interractions specifically?

Hi All, am trying to deploy a flow.. but instead of deploying the flow it is running it..am a little confused. could anyone help me? below is the statement:

prefect deployment build ./data_source_flows/master_flow.py:master_flow -n master_flow -q test --storage-block github/flows --apply

Hi, I've looked through a couple questions on the helm/k8s setup for Orion but havent found my answer. This is probably just me missing something about k8s itself. Im just trying to test self-hosting orion on my local machine using the helm chart found here and then opening the locally running dashboard. The command im using is

helm install prefect-orion-poc prefect/prefect-orion --values  helm-orion-values.yaml

and my values file is

namespaceOverride: helm-prefect
postgresql:
  auth:
    password: password
ingress:
  enabled: true

It outputs

Get the application URL by running these commands: <http://prefect.local/>

but that comes back as an unknown host. Im pretty new to K8S so any help is appreciated.

Can someone help me with this question? https://stackoverflow.com/questions/74757050/how-to-deploy-locally-multiple-flows-using-prefect-2-0

Hello, does anyone use Prefect Helm Chart? https://github.com/PrefectHQ/prefect-helm#the-database-deploys-correctly-but-other-services-fail-with-bad-password I think I have this problem but I am not sure I understood the solution

Hey 🙂 I have shared code that is used by both prefect flows + system where no prefect is installed. Thus I need a logger which is able to log everywhere. I created an own logger but have some problems with prefect to get it running. I only found documentation on how to set up a custom logger for v1: https://docs-v1.prefect.io/core/concepts/logging.html#extra-loggers But I think it should work the same for v2. I also had a look here: https://discourse.prefect.io/t/prefect-logging-faq/1476

Hey, I'm running prefect orion on Docker. The prefect UI launched fine but I'm keeping getting this error. What am I doing wrong?

Hi! I'm using Prefect 1.2.4 with docker storage and kubernetes agent I need to change the umask of the flows as they are writting to shared storage. I tried setting it in the .bashrc as I did for my users outside of flows, but apparently it is ready only for interractive sessions so not with flows! I see the recommended way is to set the umask in the entrypoint of the docker, example :

ENTRYPOINT ["./set_umask.sh"]

https://codeyarns.com/tech/2017-07-21-how-to-set-umask-for-docker-container.html#gsc.tab=0 Is there a way to override the entrypoint of the flow ? Or any other solution to set umask?

Hi all. I'm using run_deployment on existing deployments (Prefect 2.7.1) but I notice a 5 to 15 sec delay in the start of the flow run. I dont use scheduled runs so it should start right away. Should I look for parameters or config to remove the delay or is it something enviroment related?

Hi I was wondering if there are any guidelines/ideas on running subflows. As far I know we can do it either; by referencing/calling them from your code directly or using

run_deployment

. Assuming that all prefect flows are contained in the same repository, and are deployed to the same orion server. Are there any benefits/disadvantages over one or the other?

I’m attempting to set up Prefect 2.0 and I want to run flows in a docker container. My docker image is in an Azure Registry. I created a block referring to the image. I’m getting the following error: No module named prefect.engine.__main__; ‘prefect.engine’ is a package and cannot be directly executed. Any clue what I’m misssing?

I have got below error while running my code in Prefect cloud. The flow generated large number of tasks, so I am assuming there is a limit for number of tasks per flow?

12:48:12.659 | ERROR | prefect.infrastructure.process - Process 'optimal-toucanet' exited with status code: -9; This indicates that the process exited due to a SIGKILL signal. Typically, this is caused by high memory usage causing the operating system to terminate the process.

Hello, I’m reposting an issue i’m having as I have not been able to resolve. I have created a deployment using an ECStask block. The ECStask block is configured to run an existing task definition. When I try to run the deployment I get the following error: Submission failed. KeyError: "No class found for dispatch key 'ecs-task' in registry for type 'Block'." This error shows up under details of the flow run but there are no other logs. I already tried registering the block again. cc: @Taylor Curran

Hello, I have 4 tasks that run successfully when running locally using FlowRunner. When running in Prefect Cloud, I get the following results How can I troubleshoot this? • TypeError("'generator' object is not callable") • SystemError('unknown opcode') • SystemError('Objects/dictobject.c:1527: bad argument to internal function') • The 4th task hangs indefinitely

I’d like to follow a workflow with Prefect 2.0 of separate

@task

functions called from a

@flow

- all of our work is done outside of Python called via make commands, which I currently use

shell_run_command

for. However, you can’t put

shell_run_command

within a function decorated with

@task

. I could just remove the

@task

decorator. Is there a workflow folks use to leverage

@task

decorator that run shell commands?

Hi...I need help printing task results using Prefect 1: seems like printing the output only prints the task name, not task result:

from prefect import task, Flow

@task
def get_value():
    return 10

with Flow("task-results") as flow:
    v = get_value()
    print(v)

state = flow.run()

I get

<Task: get_value>

instead of

10

. What am I doing wrong?

I haven’t been able to find info on this yet, but I’m guessing it exists in docs somewhere: file structure. AFAICT

prefect deployment build

only allows for pointing at one file. If you want to pull in imports from other files like utility functions (not pypi packages) how do you suggest doing that? I assume we don’t have to have all code in the one file the deployment build command points at?