Is it possible to run a task B on a schedule, but also iterate through a list returned from an upstream task A? So for each rerun of given task B, it will iterate through the list from task A one iterable at a time

Hi - I’m trying to switch from Pickle storage to Script storage for my flow (in an attempt to work around an issue with local code dependencies). Having done this it uploads successfully, but then at run I get an error

Failed to load and execute Flow's environment: ValueError('No flows found in file.')

Why is this?

Hi All, I seem to be stuck with this problem https://github.com/PrefectHQ/prefect/issues/4393#issuecomment-927268724. I have bash installed and the shell path set

I am trying to install prefect with following command

pip install "prefect[viz, snowflake, azure, kubernetes]" and it is taking more than 3 hours and still not installed. I am continiously getting following messages

Hi :) I have a question about the flow-hash. When I change a method-signature from a method i use in a task, the hash of the flow is not changing. This results in a not reregistered flow if I use it with S3. More details are in the thread.

Hi, I have a few custom parameters that take in a user supplied string and apply some standardisation (in the run method). However, result / target templating is still seeing the original string. Is there a way to use the transformed parameter instead?

Hi I have a question: I'm using my laptop to code the flow, then I deploy it. After that I open my vps and run as Agent and seem like it not work

So my question is, how can I compile my flow so it can run on any agent? Does Prefect support that?

Hey folks 🙂 With Prefect 0.15.6 we are experiencing a (weird?) behaviour, which we didn’t have before. We have a flow, with no schedule defined. When we register the flow, Prefect activate the schedule. As a result, we see this in the Prefect UI

Hi @Kevin Kho , all task is almost functional, so then flow in general, in functional. So my question is why not support to deploy Agent on serverless function like AWSLambda, Cloud function,...etc?

If I am trying to download N # of files from S3, load them into SQLServer, and then run a DBT job - am I better off using LOOP or Map?

Hey everyone 🙂 I have trouble with triggering flow inside flow I have a flow with (I'm shortening):

with Flow(

flow_name,

run_config=run_config_provider(flow_name),

result=results_provider(os.environ["GCP_PROJECT"]),

storage=storage_provider(flow_name),

) as flow:

portfolio = fetch_portfolio(portfolio_uuid)

parser_results, results_log, csv_log, *input_row_size* = run_parser(

portfolio, excel_bucket_url, workflow_id

)

.... more tasks...

# run the next flow

with case(parser_results, "completed"):

start_flow_run = StartFlowRun(

flow_name=automation_flow_name,

project_name="project_name",

)

start_flow_run(

upstream_tasks=[updated, *input_row_size*],

parameters={

"portfolio_uuid": portfolio_uuid,

"input_row_size": *input_row_size*,

},

)

portfolio_uuid = string input_row_size = int When I run the flow it fails when attempting to run the new flow here is the error message (in the message thread): What did I do that was wrong?

How to authenticate for S3Result() if my AWS_CREDENTIALS are stored in prefect cloud secrets?

Is there a Prefect book? Would love something deeper than a basic tutorials but more than just documentation

Hello Prefect community! I would like to be able to run a flow using the

DaskExecutor

and pass a parameter like

num_dask_workers

which would be used when initializing the

DaskExecutor

at run-time. Is it possible to parametrize a flows Executor properties?

Hi all, very new to Prefect. Question about using existing functions. I already have a function in an internal package for posting messages on our internal IM, and import that function. If I want to use it as a task what is the best way to do that? Let's say I have a function postMessage(message) is this right?

import prefect

from foo import postMessage

@task

def postNewMessage(message):

postMessage(message = message)

with Flow("My Flow") as flow:

postNewMessage(message = "Hello World")

or is there a way I can just use my existing function as task instead of nesting it?

Hi, we started exploring the prefect from last week, was curious is there an example of dockerfile and docker-compose to spin up the prefect in docker, so whenever we do like a

docker-compose build

it will make prefect available at 8080 localhost, instead of spinning up the prefect locally with following steps

pip3 install "prefect[GitHub,kubernetes,Snowflake]"

Make sure you have Docker installed before starting up prefect server.

1.

Set Backend to Server Instead of Cloud - prefect backend server

2.

Start Prefect Server - prefect server start

3.

Visit <http://localhost:8080>

anyone have any idea why my task run concurrency settings would be getting ignored? these tasks started around the same time (within 10 mins) but not like the same nanosecond or anything

Hi everyone! I have configured an Automation in the prefect UI to send slack messages when the Flow state changes. When I am registering a new flow, is there a way of linking that new flow with the

automation action

without going to the UI?

Hi community, is there a way to change the flow run context during a task run, so that the subsequent tasks can see the updated context values?

Using

DockerRun

or

VertexRun

(I'm using prefect from master branch), is it possible to specify an image elsewhere than DockerHub (along with secrets to access it)?

Hello! Hope you are doing well and are motivated enough to help with an issue/question. My workflow consists of both map and reduce tasks. The mapped tasks have the split workload, so 2GB memory per worker is fine, but then when I do a reduce, I would like for the worker on this particular task to have more memory than an averaged worker. I tried increasing the memory limits in dask, but this works for all the workers, while I would like to specify the custom increased limits only on the worker for a specific task. Is this possible? Thanks a lot!

I'd love to know what everyone is currently doing for event based flows (given that https://docs.prefect.io/core/PINs/PIN-14-Listener-Flows-2.html) is currently paused. I have a massive home grown job system that spans multiple runtimes all inside Kubernetes jobs that I would like to start porting to something like prefect so we can get a better handle on reliability and visibility. What I need to be able to do is start flows in response to end-user config changes. Ideally there is also a way to throttle or debounce based on parameters.

I am trying to get the

README

of a flow through a graphql call. I am able to do this by calling

flow_group.description

but: • What is a flow group? Is that all versions of a given flow? • If so, why does

flow_group.name

return a uuid? (aka. how can I simply relate a flow’s name to it’s description)

I’m currently receiving user input through

model = Parameter('model_name')

. I want to announce the user input in Slack via my state handler, but when I do it displays as

<Parameter: model_name>

rather than the actual user input. Does anyone have any suggestions on what I can do to display the user input?

I have a task within a flow that returns a structure

val = my_task()

where val is some json thing. I also want to set that tasks downstream dependencies, though, which I had been doing with

val.set_downstream(othertask)

(although I was always a little confused as to why that would work). Well, now it's stopped working. What's the right way to do both of these things?

Got an Orion question: Will I be able to map subflows across a Dask cluster?

I'm seeing some odd behavior with mapped tasks and calling AWS Batch jobs. Essentially I have one mapped task that kicks off N batch jobs and returns a list of batch job ids. The next mapped task takes those job ids as input and runs

AWSClientWait

to monitor the status of each of those jobs. Here's the code for that mapped task -

@task
def wait_batch(job_id, delay, max_attempts):

    logger = prefect.context.get('logger')

    <http://logger.info|logger.info>(f"Waiting for job to complete: {job_id}")

    waiter = AWSClientWait(
        client='batch',
        waiter_name='JobComplete',
    )
    waiter.run(
        waiter_kwargs={
            'jobs': [job_id],
            'WaiterConfig': {
                'Delay': delay,
                'MaxAttempts': max_attempts
            }
        },   
    )

    <http://logger.info|logger.info>(f"Job complete: {job_id}")

    return job_id

But what we are sometimes seeing are one or more batch jobs failing, which then somehow stops the other jobs from responding to this AWSClientWait call... and so the mapped task keeps running even though all the jobs have either failed or completed. Any ideas?

I was running a prefect flow but facing a run out of memory issue, how can I clear up and free the space. I saw a 17G .prefect file in the root directory, can I remove it?