Hello Prefect community! Just a quick question: is there a simple convenience function to check if a flow has been successfully registered? (e.g.

flow.is_registered()

). Otherwise, is there a gist for using

prefect.client

and working with the graphql response? Thanks in advance

Quick question. Is there a simple way to run a flow via an API endpoint rather than scheduled within prefect?

Hi! I am trying to call a task inside a for-loop, but keep getting the following error:

TypeError: Task is not iterable.

What is the recommended way to go about this?

Hi ! As anyone tried to validate the parameters of a flow using a Pydandic model or dataclasses ?

Hi all 👋 I was wondering if there is something like a Mattermost notification task? I figured as there is a slack notifier something for Mattermost might be not so far off? Cheers, C

Does anyone have an example of rendering a matplotlib graph with the artifacts api?

We’re using the k8s agent with dask executor on GKE with workload identities. Does anyone know if it is still required to set the

GCP_CREDENTIALS

secret in Prefect Cloud? Or will it just use the service account tied to the namespace? The latter would be nice to avoid creating any long lived service account keys.

Hey folks, I'm trying to do following but can't figure it out. Any kind of pointers will be appreciated. I want to build a pipeline where it loads a file larger than RAM my local machine have. I can do that with Dask alone, but not sure how to do that in prefect task. In other words, is it possible that the first task of a flow load a chunk of dataframe and once it processed, it will load next chunk and so on? Thanks.

Why are our flow limited to a concurrency of 2 tasks? We are using prefect cloud with a Kubernetes Agent running 2 workers. For some reason it only runs 2 tasks concurrently. I have tried setting the executor to DaskExecutor() and LocalDaskExecutor(). The behaviour stays the same The tasks I am running in this example are just waiting for 10 seconds, so resource constraints can’t really be the issue. I assume it’s limiting it to 1 concurrent task per Kubernetes worker node, but haven’t been able to confirm that. Anyone know any possible reasons for this?

Hey there. We'd like to preserve all the prefect logs as files in a file system. How can we get all prefect logs to be written to files as well? Thanks!

Hi all, a stupid question. I was deploying a Prefect Server on a remote machine - are there any settings needed to be able to view the UI from the local machine?

Hi all, I have a quick question about state handlers. I’ve created one for the use case if the flow ends up in a failed state, I want to grab all the error details from underlying tasks and send an email. What I’m doing when testing locally is:

for t in flow.tasks:
        tr = state.result.get(t, None)
        if not tr: continue
        <http://flow.logger.info|flow.logger.info>(f"Task name '{t.name}' -> State '{tr.message}' -> Is failed: {tr.is_failed()}")
        if tr.is_failed():
            err_params['tasks'].append(OrderedDict({
                'taskName': t.name,
                'errorMessage': tr.message,
                'errorParams': tr.result if isinstance(tr.result, dict) else str(tr.result)
            }))

Basically just going through all the tasks and checking if they are is_failed is true, if so I am grabbing some info. This all works as intended locally, but when I package my flows in a docker and run with docker agent, it turns out state.result is an empty dictionary. While previously locally, I found each tasks result there. I’ve played with GCSResult and just leaving it to default but in both cases while running in docker the state.result={}. Any ideas what I might be doing wrong?

I am trying to run a flow via an ECS agent. The agent is up and running, but when I trigger a flow I get the following error:

Parameter validation failed:
Unknown parameter in input: "null", must be one of: capacityProviderStrategy, cluster, count, enableECSManagedTags, enableExecuteCommand, group, launchType, networkConfiguration, overrides, placementConstraints, placementStrategy, platformVersion, propagateTags, referenceId, startedBy, tags, taskDefinition

This is coming from the boto3 call triggered here: https://github.com/PrefectHQ/prefect/blob/6b59d989dec33aad8c62ea2476fee519c32f5c63/src/prefect/agent/ecs/agent.py#L320 My agent is being started like so (with prefecthq/prefect:0.14.13-python3.8, maybe this is old?):

prefect agent ecs start --run-task-kwargs s3://$bucket/run_task_kwargs.yml -a <https://my-api:4200/graphql>

My run task kwargs yaml has

cluster

,

launchType

and

networkConfiguration

. I also attempted to provide values in the run config via the UI, but it did not seem to change much (added CPU, memory, task and exec role). Any help is much appreciated!

hey all, quick question, is it possible to configure task output names for tasks with nout > 1 ?

How do I deal with the following error?

[2021-07-29 12:41:09-0500] INFO - prefect.LocalDaskExecutor | Attempting to interrupt and cancel all running tasks...

question about memory management - I have a flow along the lines of:

with Flow() as f:
   big_obj = gen_obj1()
   big_obj2 = gen_obj2(big_obj) #never need to use big_obj again, want to reclaim memory
   big_obj3 = gen_obj3(big_obj2)    # finished with big_obj2
   ...etc

what's the canonical way of freeing up these objects within the flow, once they've been consumed and are no longer needed?

Scheduling issue... Running basic task

hello_world.py

the Prefect.io server runs on a VM on AWS, with python 3.6 with latest prefect python package agent is running: agent LocalAgent LAST QUERY 11:19am | 1 seconds ago CORE VERSION 0.15.3 when running

hello_world.py

using python3.6, the schedule does not work when running using python3.8, the schedule works, but I get errors Why does the schedule does not work with the python3.6 ? (It used to work.. not sure what changed) I tried pip install prefect --upgrade (both on laptop and VM) killed the old agent (on lap top) prefect backend server (on laptop) prefect agent local start --label "<label>" (on laptop) this did not help

What is the typical prefect version upgrade strategy for a cloud KubernetsAgent?

In the functional flow context, how does one use

.set_upstream

to create a relationship between two tasks, where the downstream task doesn't care about the upstream's (in-band) results? The docs say it's possible, but I end up with two copies of the downstream task when doing what seems like the obvious.

Hey guys, I know there is a Task retry logic that you can add ( on a failure ) but is there a Flow level retry?

Hi, I had a question about script and module based storage - under what circumstances does a flow need to be re-registered with script/module based storage when the underlying code changes? Is it only when the structure of the Dag changes?

Question about retries If I have a sequence of independent tasks, is it possible run them, and then rerun only those that failed ? but without waiting for the task that failed to rerun. only at the end of all the tasks get back to those that failed ?

Hello, I am just testing a Docker recipe which contains the docker command

RUN prefect auth login --key ${PREFECT__CLOUD__API_KEY}

where `PREFECT__CLOUD__API_KEY`` is an ARG. But I'm getting this error:

#13 1.047 Your API key is set in the Prefect config instead of with the CLI. To log in with the CLI, remove the config key `prefect.cloud.api_key`

HI! I’m trying to convert existing ETL pipeline into prefect so I’m using prefect cloud as a server and using one of my ec2 server as an agent. On the ec2 agent I registered the existing job as flow and task and ran flow.register() to register on my server on the etl task, I do have a custom class that I import from local path and it seems like when i run the task from prefect ui, it can’t refer from the local python class. Following is the error message

Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'common\'",)\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.',)

Again thank you so much!

Hi, is there a way to use the github flow storage backend not only to store and retrieve the flow to run, but also some common code shared between flows, like a file commons.py? ... if I deploy the commons.py on agent side, it can be imported, but that restricts me to one version, whereas the flows can be always the newest version from a git branch. thx

hey just wondering if there is a way to complete a complete chain of mapped tasks before moving on to the next iteration?

files = [1, 2, 3]
    data = extract_data.map(files)
    upload = upload.map(data)

I would like to transform the above code so that file 1 will be extracted and uploaded before moving on to file 2. I know i could wrap this functionality in to one function but i would prefer to keep them separate

Does anyone have a working example of using

BatchSubmit

and

AWSClientWait

in order to submit a job and wait for it?

with Flow('flow'):

    batch_input = Parameter('batch_input')

    job_res = BatchSubmit(
        job_name='job_name',
        job_definition='job-def',
        job_queue='job-queue',
    ).run(
        batch_kwargs={
            'parameters': {
                'batch_input': batch_input,
            },
        },
    )

    wait_res = AWSClientWait(
        client='batch',
        waiter_name='JobComplete',
    ).run(
        waiter_kwargs={
            'jobs': [job_res]
        }
    )

This is what I have. Feels a bit weird to call

.run

directly on the task (I am new at this and have mostly just run

@task

annotated functions). Additionally, boto is complaining that the

batch_input

values need to be strings:

type: <class 'prefect.core.parameter.Parameter'>, valid types: <class 'str'>

Hi there, try to get a flow that looks like this to run, currently

test_reduce

gets skipped but I would like for it to receive

[1, 2, 4, 5]