Can some one tell me why my flow is getting picked up by 2 agents?

Another issue I came across is when you combine nout > 1 and .map: The multiple return values from the nout>1 task are not iterated correctly when passed as args to the next task

from typing import Tuple

from prefect import Flow, Task, task


@task(nout=2)
def step1(num) -> Tuple[int, str]:
    chars = ['a','b','c']
    return num, chars[num - 1]

@task
def step2(arg1, arg2):
    print(f"arg1={arg1} arg2={arg2}")

def unzip(nout: int) -> Task:
    @task(nout=nout)
    def _unzip(res) -> tuple:
        return tuple(zip(*res))
    return _unzip


with Flow("nout test") as flow:
    numbers = [1, 2, 3]
    # bad
    a, b = step1.map(num=numbers)
    step2.map(arg1=a, arg2=b)
    # prints:
    # arg1=1 arg2=2
    # arg1=a arg2=b

    # good - with unzip workaround
    c, d = unzip(2)(step1.map(num=numbers))
    step2.map(arg1=c, arg2=d)
    # prints:
    # arg1=1 arg2=a
    # arg1=2 arg2=b
    # arg1=3 arg2=c


if __name__ == "__main__":
    flow.run()

Hi team! Is there an easy way (preferably in the UI) to cancel all running flows (anything in the blue state)?

Good morning, I have a question about flow registration. I'd like to register my flows during CI/CD (specifically, AWS Codebuild). My issue is that I have run into import errors, because my flow dependencies are not installed in the build environment. I am using

S3Storage

and

KubernetesRun

. Thanks in advance for any guidance.

Hi folks, I've created a flow that uses Docker Storage, and an ECS run config. The run config has a task definition defined locally in a YAML file. I am using Prefect Cloud as my backend. The flow registers ok but when I attempt to run it the task definition cannot be registered because I have not specified a image to the ECS Run Config. However in the API reference it says: • `image (str, optional)`: The image to use for this task. If not provided, will be either inferred from the flow's storage (if using 

Docker

 storage), or use the default configured on the agent. and I am using Docker Storage, which has worked in the past and does get stored in my image repository (ECR). It seems as though the specfiying an explicit task definition, either through the

task_definition_path

or

task_definition

arguments and using Docker Storage does not work, or there is a bug in the image being picked up from Storage. Any guidance would be welcome.

Do Agents submit separate flow runs to the Executor in a depth-first or breadth-first pattern? I’m wondering if Prefect gives priority to task B of the 1st flow run (“1B”) over the task A of the 2nd flow run (“2A”).

Hi folks, what is the current approach for specifying a custom prefect context (to be used as a default) when registering a flow ?

Hi folks, we spotted a change in

0.14.11

that made the Flow object no longer pickleable using

pickle

- I am wondering if others have bumped into this

Hello everyone, is there a way to set the state of a task to Skipped, after all retries have failed ?

Question about using secrets. I have my flow running successfully ... I would like to know if this is the best practice on using secrets: • create them outside the flow • calling the .run() method on them

aurora_user_val = EnvVarSecret("AURORA_USERNAME", raise_if_missing=True).run()
aurora_pass_val = EnvVarSecret("AURORA_PASSWORD", raise_if_missing=True).run()
aurora_host_val = EnvVarSecret("AURORA_HOST", raise_if_missing=True).run()

with Flow("test_vars") as flow:

Hi all, looking for some help debugging here. I have a task which finishes fine when run locally, the logs indicate it finished fine when run through an agent, but the task itself is stuck in Running. Its a simple flow (it turns a tiny pandas dataframe into a json object), and I make a call to the logger directly before the return statement. I see my log statement that it all worked, but the task state doesnt change. Is there a good way of debugging what on earth might be going on? The odd thing is that the task it never finishes is a common task (I download 4 things and process them all in the same way). 2 of them finish, every time. 2 of them get stuck on running, without fail. EDIT: I seem to have found a potential bug in Prefect. These hang ups occur when I have

@task(timeout=20)

, but they just completed successfully with a normal

@task

annotation. Will update prefect now and check to see if that helps

Trying to make a task decorator that subsamples and caches the output of a task to HDFS (each task returns a spark dataframe). The goal is to quickly iterate on and debug downstream tasks using subsampled data. Since these are not full blown checkpoints, I'm not sure if the Results API would be appropriate. I was thinking something like

@cache(subsample=100, sdf_key='sdf_large')
@task
def some_large_spark_dataframe():
  "intensive ETL process here"
  ...
  return sdf


@task
def downstream_task(sdf_large):
  "some intensive computation on sdf_large"
  ...
  return sdf


with Flow() as flow:
  sdf_large_sample = read_from_cache('sdf_large')
  downstream_task(sdf_large_sample)

...but I've had difficulty stacking decorators with the

task

decorator. Moreover, the usual challenges of the task result not being available until evaluation time. Any recipes you'd recommend?

Hi all, I'm trying to have dependent flows, where the parent flow that needs to run only once. Consider the following scenario: 1. Parent flow is an expansive process that preprocesses data and loads into a storage (let's say S3). Ideally, it shouldn't run more than once. 1. There can be multiple children flows that use the preprocessed data. I've taken a look at this. However, with this setup, in order to ensure for the parent flow to run only once, I need to put all children flows in a single "flow of flows". However, I envision a use case where different teams bring in their own flow and use the common preprocessed data. My understanding is that if each team's flow define the parent flow as dependency using

upstream_tasks

, it would cause the parent flow to run multiple times. I think what I need is like the factory pattern in OO paradigm. In Airflow, I think I can use ExternalTaskSensor to achieve this. - Does my question/use case make sense? - Is there a preferred "Prefect" way to solve this? - Or should I keep the state of the parent job externally?

What happens if the agent dies in the middle of a flow? Will the flow continue when a new agent comes online or will the flow error out?

Hey everyone, I'm really struggling to wrap a custom Executor into Prefect and could use some help! I know it's a little late in DC, so let me know if this should wait til tomorrow lol. So I have

future = CustomExecutor(flow.run, *my_args, **my_kwargs)

working just fine. The issue comes when I try

flow.executor = CustomExecutor(); state = flow.run()

. When I do this with, I recieve...

Unexpected error occured in FlowRunner: AttributeError("'CustomFuture' object has no attribute 'is_mapped'")

I read through the FlowRunner code and have a good guess as to what's happening, but I could use some insight from a dev. Has anyone seen this before?

Hi. I have a flow where I get a token for an api that goes stale after some known time. Is there a way to indicate that a depending task should be rerun. An example of the flow could me as follows:

from prefect import task, Flow, unmapped
import requests

def task_that_goes_stale():
    token = code_to_get_token_at_some_cost()
    return token 

def task_using_token(user,token):
    r = requests.get('<https://api.github.com/user/{user}>', headers = {'TOKEN': token})
    
with Flow("my_flow") as my_flow:
    all_users = [i for i in range(1000)]
    token = task_that_goes_stale()
    task_using_token.map(all_users,unmapped(token))

I know i could just get a new token for each call but this seems like the wrong way to do it. Is there a Prefect way to do it? I can see that there is output caching but as I read it this is meant for bringing results between flows https://docs.prefect.io/core/concepts/persistence.html#output-caching

Hey Everyone, Im currently trying to get a better grasp on how to best interact with past results of flows. Something similar to what metaflow has https://docs.metaflow.org/metaflow/tagging would be great. Is there some preferred way to do this? Something like:

with Flow("A Flow") as my_flow:
    task_res = my_task()
task.run()

followed by:

latest_flow_state = my_flow.last_successful_run
latest_flow_state.tasks[task_res]

Hey, I am using Prefect Cloud on version

0.13.18

. Has anyone had this before -- all of a sudden a large row of cancelled flows appear in the UI? (these flows run using the CronSchedule)

After upgrading to v0.14.11 I can no longer start my local stack. Any guidance on resolving this?

$ prefect server start
ERROR: The Compose file './docker-compose.yml' is invalid because:
services.ui.depends_on contains an invalid type, it should be an array
services.apollo.depends_on contains an invalid type, it should be an array
services.towel.depends_on contains an invalid type, it should be an array
services.graphql.depends_on contains an invalid type, it should be an array
services.hasura.depends_on contains an invalid type, it should be an array
Exception caught; killing services (press ctrl-C to force)
ERROR: The Compose file './docker-compose.yml' is invalid because:
services.ui.depends_on contains an invalid type, it should be an array
services.apollo.depends_on contains an invalid type, it should be an array
services.towel.depends_on contains an invalid type, it should be an array
services.graphql.depends_on contains an invalid type, it should be an array
services.hasura.depends_on contains an invalid type, it should be an array

Does the EmailTask’s

email_to_cc

only allow for sending one email for cc?

with a local agent and local executor, is there an easy way to keep track of the location of an external file? i'm trying to do spark-submit with the

ShellTask

, but it needs the location of the file i'm submitting with. Right now, I've hardcoded the location in my file, and switching to that location with

helper_script

option on

ShellTask

when using the

GITHUB Storage

if you are referencing an outside file that is in the same repository, is the flow able to find and run it?

Hi, I am attempting a KubernetesRun using DigitalOcean’s managed kubernetes service. I am using S3 for flow storage, and a public Docker image to run the flow. Kubernetes agent has been registered, and S3 credentials specified as env vars following the docs :

prefect agent kubernetes install -t token_value -l label -e aws_access_key_id=XXX -e aws_secret_access_key=YYY --rbac | kubectl apply -f -

The agent is installed and running in the kube cluster on inspection using kubectl get pods

NAME                             READY   STATUS    RESTARTS   AGE
prefect-agent-58cf5b46d5-84hh6   1/1     Running   0          26m

Running the flow in Prefect server, and the flow gets picked up by the agent, but encountering the error:

Failed to load and execute Flow's environment: NoCredentialsError('Unable to locate credentials')

This is despite the agent having knowledge of the S3 creds via the -e flags. I have also tried passing the S3 creds as a dict to KubernetesRun inside my flow:

KubernetesRun(
  image="my_image",
  labels=["my_labels"],
  env={
    "aws_access_key_id": "xxx",
    "aws_secret_access_key": "yyy"
  }
)

Am I missing something for my kube cluster to authenticate to S3?

Does anyone know how to debug a Kubernetes run with the error:

Pod prefect-job-7b9f3bf6-z4p27 failed.
	Container 'flow' state: terminated
		Exit Code:: 1
		Reason: Error

Hi all, I am really struggling getting this to work. I am trying to use

Docker Storage

with a

ECSRun Config

. What I am looking to make happen is to have the github repo cloned to the docker container so that my flow has access to various files (jupyter notebooks primarily). I have been trying solutions for a number of days and I am currently stuck on

Error while fetching server API version: {0}'.format(e)

docker.errors.DockerException: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or direct

This would indicate that my prefect image does not have access to the docker deamon. But I can’t figure out what I am doing wrong. I have

prefect backend cloud

set. And below are my files.

Hello! I am setting up my Prefect flows CI/CD pipeline, and as part of that I need to register flows against my Prefect server instance. One thing I noticed is that there's a

set_schedule_active

parameter that can be specified to the

.register()

call in order to automatically enable or disable the scheduling of the new flow version. These values can be either

True

(enable the scheduling) or

False

(disable the scheduling), but there's no option to "leave it as it is right now", meaning that if the flow is active/inactive, the new registration should leave the new workflow as active/inactive as well. What's the best approach in this scenario? I want to avoid 2 potentially dangerous situations: - Accidentally enabling a flow that should remain disabled - Accidentally disabling a flow that should remain enabled Avoiding any specification in the code would be ideal, since we don't want to enable or disable flows using the code, but rather the UI/API. Thanks!

hey folks—i’m new to Prefect and am trying to understand how one might approach configuring a flow so that it has access to its own previous execution date (similar to airflow’s

prev_execution_date_success

variable). in reading the docs, caching doesn’t quite seem like the right fit. persisting output makes sense, but not clear to me the right way to re-consume that output on a future flow run. so at this point my inclination is to simply read/write the last run timestamp to a local file. anyone have a recipe, docs, or otherwise on a better approach?

Hi Team, I’m using Prefect core, and would like to track identity of user who created/updated a flow. Any suggestion about how to do that properly?

Thanks