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]

Hi, can I get an explanation for what causes this versioning error? Where are the two environments mentioned here?

Failed to load and execute Flow's environment: FlowStorageError("An error occurred while unpickling the flow:\n  TypeError('an integer is required (got type bytes)')\nThis may be due to one of the following version mismatches between the flow build and execution environments:\n  - python: (flow built with '3.7.5', currently running with '3.8.11')")

Looking for help implementing a simple flow of flows. I am creating a parent flow using two registered flows, A and B. A is upstream of B. Flows A and B have parameters so I use Parameter tasks in the parent. Ultimately I want to pass results from A to B but still trying to get simple A then B with params running. I am trying two solutions: One with StartFlowRun and the other with the new create_flow_run. With StartFlowRun I can create a parent that will run A then B, using wait=True, but dont understand how to pass parameters or the result from A to B as a parameter. With create_flow_run, I am trying the following: create_flow_run A with parameters, wait_for_flow_run A, create_flow_run B with parameters, wait_for_flow_run B. When I create a parent flow in that order, it runs A and B at the same time. Looks like I am missing the way to specify A is upstream from B and wait until A is done before running B. I am also having trouble with get_task_run_result for task A using the return id from the call to create_flow_run A. Any help appreciated.

A couple things: Is there some lag between editing

config.toml

and secrets added to it becoming available when doing local testing? That appears to be the case, since a missing local secret error went away seemingly magically, as I was preparing to ask about it. Alternatively, I had added the secret to cloud first, so maybe there was a (longer) lag before it found it there, and it's still not using my local-secrets? Also, SlackTask doesn't work when supplying 'message' at runtime (or directly to run); I get

TypeError: run() got multiple values for argument 'message'

- something to do with that defaults_from_attrs magic?

Hi everyone! I have a testing question: How do I mock method calls inside a Task? This is my current setup: File 1:

def do_something_1(params_1):
  (does something)

File 2:

from path_to_file.file_1 import do_something_1

class SomethingTask(Task):
  def run(self, params_1):
    do_something_1(params_1)
    do_something_2

File 3:

import unittest
from unittest.mock import patch
from path_to_file.file_2.SomethingTask import SomethingTask

@patch('path_to_file.file_1.do_something_1')
class TestSomethingTask(unittest.TestCase):
  def test_1(self, mock_do_something_1):
    task = SomethingTask()
    task.run()
    self.assertTrue(mock_do_something_1.called)

Instead of mocking

do_something_1()

it actually calls

do_something_1()

, and strangely the assertion returns False.

Has anyone seen the following error:

An error occurred (InvalidParameterException) when calling the RunTask operation: Task definition does not support launch_type FARGATE.

I am running a flow via an ECS agent. It worked just fine until I specified a custom value for

image

when submitting the job via the UI. The image I specified was for an ECR image