Hey guys, I have a main flow that runs dependent flows defined as attached in the file. All the flows are actually legacy Java pipeline running in containers using ECS Fargate, so task defined in this graph will trigger the job to run on ECS fargate with all dependencies packaged within each container. The execution on this main flow runs fine but it is serialized, the point of this is to run in parallel through the DAG as each task is an independent ECS task running. Is there something I am missing to execute the tasks/flows in parallel leveraging ECS?

Hi Everyone, is it possible to share variables between the task and the state handlers. i want to send slack notifications based on my config parameter. the config is a dict which i want to use within the state handler post_to_slack function to send custom notifications.

Hi folks simple smile We are using ECSRun + Docker storage. We’d like to provide the task definition ARN, but we get the following error:

Cannot provide `task_definition_arn` when using `Docker` storage

Looking at the code at https://github.com/PrefectHQ/prefect/blob/master/src/prefect/agent/ecs/agent.py it seems a by design choice. What are the alternatives for running a flow on ECS while providing the task definition ARN?

Question: I’m setting my flow to use the LocalDaskExecutor and deploying to a Docker agent. When I use a docker agent on my laptop it parallelizes just fine, but when I try to run on a VM (16 CPUs, Google Compute Engine) it’s only using a single core, and only executing one task at a time. Not sure if I’m setting it up wrong, any ideas what might be happening?

Hi all, I'm using the ECS Agent and running the tasks using the [ECS run config](https://docs.prefect.io/api/latest/run_configs.html#ecsrun). I'm passing a custom task defintion to the run configuration. I've realised that the execution role is picked up from the custom task definition (this does not match the behaviour described in the docs btw; they specify that

If not provided, the default on the agent will be used (if configured).

). This does not happen for the task role, which is instead passed from the agent. This behaviour is unexpected; I would have assumed when passing a custom task definition, that the roles I have defined (task role and execution role) would not be overridden by prefect. So, I've now modified naming convention for the task role for a particular flow, to test passing it via: `task_role_arn (str, optional)`: The name or full ARN for the IAM role to use for this task. If not provided, the default on the agent will be used (if configured). The role I have passed has full S3 permissions:

"Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "s3:*",
            "Resource": "*"
        }
    ]
}

The flow itself uses the [S3Upload](https://docs.prefect.io/api/latest/tasks/aws.html#s3upload) action. It fails with

Error uploading to S3: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied

Sorry if this has already been answered - I just couldn't find it in the docs - perhaps I've just had a really long day and can't think of the right search terms. What's the best practice for forcing a failure if a task or flow doesn't pass a custom validation I've made? For example, if the rows loaded into a destination don't match the row count that was supposed to be loaded? Stuff like that happens with Salesforce all the time and their callbacks aren't very flexible, so I like to check to see if rows loaded are what I expected.

Hi all, we're running Prefect on Heroku. We've got some flows which share state between tasks. Namely, the first task goes and partitions a set of ids by those which are still valid or not (by calling out to a 3rd party API). The second task then processes the invalid ones to determine what's happened to them. The first task returns a tuple of sets of ints. There are at most 1M ints between the two sets. At the moment it's taking multiple hours to return from the first task and mark it completed and I wondered if anyone had any advice on how to deal with this and whether this is expected behaviour?

Hi all. I am completely new to prefect and I am facing an issue. I have a prefect docker agent deployed. A flow attached to it shall be triggered when an input is passed and the flow shall be immediately set to run state. It's been working perfectly fine until yesterday I observed the flows are in the scheduled state infinitely. SO every new input keeps adding new fllow runs but all into an infinitely scheduled state. Any inputs would be helpful. Thanks in advance.

been seeing this sporadically over the last 24 hours (maybe 1 in 10 flows?), any idea what might be up? I don't see anything about

edge_on_conflict

in the

prefect

or

server

repos

File "/src/util/prefect.py", line 575, in register_flow
                                             flow_id = client.register(flow, project_name, idempotency_key=flow.serialized_hash())
                                           File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 1227, in register
                                             self.graphql(
                                           File "/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 569, in graphql
                                             raise ClientError(result["errors"])
                                         prefect.exceptions.ClientError: [***'path': ['register_edges'], 'message': '[***\'extensions\': ***\'path\': \'$.variableValues\', \'code\': \'validation-failed\'***, \'message\': "no such type exists in the schema: \'edge_on_conflict\'"***]', 'extensions': ***'code': 'INTERNAL_SERVER_ERROR'***]

Hello, we're observing an issue with Prefect Artifacts and mapped tasks after update to 0.15.6, any ideas how to resolve 🙏 ?

Hello community, I am chasing an intermittent bug, possibly related more to Kubernetes than to Prefect. Currently, a lot of our Prefect flows map out to several k8s jobs and do some moderately heavy computation for ~1 hour and then shut down. Our load is very burst-y, so we might scale from 1-2 nodes to ~50 down to 1-2 nodes again within the span of an hour. We are using Azure Kubernetes Service and I have tried being quite aggressive with the k8s autoscaler, setting the

scale-down-delay-after-add

to 1m and similarly the

scale-down-unneeded-time

to 1m. The issues that we are facing is sometimes the task pods fail, seemingly without reason, and logs are quite unhelpful. My hunch is that is has something to do with scaling of the cluster, potentially destroying pods/losing networking between flow and task pod in the process? We are already setting the

<http://cluster-autoscaler.kubernetes.io/safe-to-evict|cluster-autoscaler.kubernetes.io/safe-to-evict>: false

on all pods, so eviction shouldnt be the issue. Have anyone else had any experiences with k8s autoscaler settings leading to weird, intermittent task failures?

Hello everyone

Hey all, I'm having a strange problem that has me stumped. A flow that I run daily has just started failing. Basically the flow just launches several sub-flows using

create_flow_run

. Today it started giving me the traceback pasted below. I commented out everything in the flow except for the very first

create_flow_run

and it's still giving me this error (see thread): Any ideas?

Does anyone know if it is possible to pass

StartFlowRun

storage information? Or a better way to store flows that need to operate in multiple environments?

Not a question but I enjoyed finding this little easter egg when having a look through the Prefect source 😆 https://github.com/PrefectHQ/prefect/blob/e699fce534f77106e52dda1f0f0b23a3f8bcdf81/src/prefect/core/flow.py#L1738

I’m trying to keep a large project organized and I’m wondering if I’m thinking about it the wrong way. I have two series of tasks currently defined as flows that are defined in seperate .py files. I was hoping to create a third flow that executed the first two and then performed a few more tasks. Is it better to think of those two upstream files as flows, or should I just organize each of those files by having them end in a final task that has all the upstream tasks as “upstream tasks”, and then import the terminal task into the master flow? (hopefully thats clear)

Hi team, whom should I contact to get one of our teammates prefect account unlocked after it got locked due to too many failed login attempts?

So I saw that we can use a dictionary of parameters as the argument of `flow.run`:

flow.run(parameters={"x": 8, "y": 9})

I wonder if there is a way we can do that with

flow.register

as well? Something similar to this:

flow.register(project_name="My Project", parameters={"x": 8, "y": 9})

I use mlflow with prefect, and I run an mlflow experiment inner a prefect task, I would how can I get logging from the mlflow run experiment to the prefect UI? I juste see in the documentation how log in the task cde, but not among a fonction inner the task there my task :

@task(log_stdout=True)
def run_mlflow(project_path, experiment):
    mlflow.projects.run(
        project_path, experiment_name=experiment,
    )

and there my mlflow experiment where I would put prefect logging :

initialize(config_path="conf", job_name="gojob")
cfg = compose(config_name="config")
print(OmegaConf.to_yaml(cfg))
traking = cfg["var"]["MLFLOW_TRACKING_URI"]
params = {}
project_path = cfg["project_path"]
experiment = cfg["experiment"]
mlflow.set_tracking_uri(traking)
mlflow.set_experiment(experiment)

# print(subprocess.run(["ls"]))
with mlflow.start_run(nested=True):

    set_env(cfg)

    get_data = mlflow.run(project_path, "process_data", experiment_name=experiment)

    train = mlflow.run(project_path, "train", experiment_name=experiment)

the logging from print() appear just in the task and not in the mlflow experiment Thank you in advance

Hi, I want to ask if Orion still uses two types of storage:

script

and

pickle

based storage as described here: https://docs.prefect.io/orchestration/flow_config/storage.html#pickle-vs-script-based-storage

Hi, I'm new to prefect. I've been trying to setup Prefect on my local machine and I'm having a difficulty understanding how to start Prefect CLI? I'm trying to run basic commands like "prefect backend cloud" but it is not recognising the "Prefect" command to begin with. Is there a documentation which I can refer to setup Prefect CLI?

Has anyone used prefect with feast on GKE?

Hi! Trying to restrict the memory usage of each flow by passing

host_config

to our

DockerRun

config:

flow.storage = Docker(
        registry_url=...,
        image_name=...,
        files={...},
        env_vars={...},
        python_dependencies=[...]
        )

client = docker.APIClient()
host_config = client.create_host_config(mem_limit=12345,
                                        mem_reservation=1234
                                        )
flow.run_config = DockerRun(host_config=host_config)

flow.executor = LocalDaskExecutor()

When registered to Cloud, this seems to be ok, since starting a Run shows the following default in Host Config:

{
  "Memory": 12345,
  "MemoryReservation": 1234
}

However, this seems to have no effect on the the newly created flow containers.

docker stats

show that MEM USAGE keeps growing and a LIMIT that equals the total server memory.

docker inspect <CONTAINER> | grep \"Memory[\"R]

gives

"Memory": 0,
            "MemoryReservation": 0,

What are we missing here?

Hello! We are trying to start building our data processing pipelines on Orion. However I stumbled upon a weird behaviour on deployments which make impossible to use Orion at all. • My flow is working perfectly and I can trigger it and see it on the UI. • When I define the

DeploymentSpec

with the location and name and register it, the output seems OK. •

prefect deployment ls

does list my deployment. But: • The UI does not show my deployment. • The command

prefect deployment inspect my_deployment

does nothing. Let me know if you need more details or logs to investigate 🙂 ! thank you by advance and have an excellent day !

I'm trying to understand the best way to scale out our tasks on GCP. It seems like there are two paradigms, either an

Agent

or

Executor

(or both). I read this SO answer already, along with this blog post on using an ECS Agent, which doesn't mention Dask. Background: our workflows are heterogenous between tasks (some just moving some stuff around, some very parallel, some requiring GPU) and between runs (sometimes will be much more parallel), and very bursty. Some tasks are very long-lived and will spawn cloud containers of their own, but I assume Prefect doesn't need to care about that (unless we wanted those to be able to log back to Prefect). VertexAgent/VertexRun Following the blog post, we could set up a Vertex

Agent

(just merged on GitHub, not on PyPI yet) as the agent. We could then set up a Flow-of-Flows (as can only specify

VertexRun

parameters at the Flow level), specifying the image, CPU etc for each flow. This would by default use a

LocalExecutor

, so any parallelism within that Flow would simply be run sequentially? But each separate run of that Flow would automatically spin up the needed instances. DaskExecutor If we wanted Task-level parallelism, we'd need to use the

DaskExecutor

? So then we'd have a Vertex instances running the

Agent

, and then a bunch of Vertex instances running `Flow`s, and then a Dask cluster running parallelised `Task`s? Have I understood that correctly? Would there be a way to avoid Dask by eg parallelising at the flow-to-flow interface?

Hi! I've mapped a task of mine, and some of them had to retry. But, after they successfully finished (the 'End Time' has been shown in the Mapped Runs tab of the task). some of them didn't change their state and are still running. Is it a bug or should I just be patient? Thanks in advance! ps.: this only happened to some of the ones that where retried.

i’ve got a task that depends on two other prior tasks, how can i get the child task to only start after both parents have finished (i wanted to combine the output of both tasks into a single arg instead of adding a new arg)? i tried using a simple boolean

and

(the output of each task is a bool), but it looks like prefect started the child task when the first parent finished instead of waiting for both:

FIRST_TASK_DONE = first_task()
SECOND_TASK_DONE = second_task()

THIRD_TASK_READY = FIRST_TASK_DONE and SECOND_TASK_DONE
third_task(third_task_ready=THIRD_TASK_READY)

third_task

was supposed to only start after both

first_task

and

second_task

had finished, but instead it only waited for the first one to complete

Ran into a strange issue while clearing my Account data. I had set up a personal Team to prove out some use cases for Prefect and then found out my manager had set up a Team already for this purpose. While joined to both Teams, I went to the Account tab on my personal Team, cleared all data, and removed myself and all users from that Team. Everything seemed to work swimmingly, until I realized that all content I had personally created in either Team was deleted. This included a test Flow and a service API key I created for AWS ECS. Furthermore, I was now the only member of the Team my manager created, seeming to mean that purging the Account data actually did so for both Teams. Is this expected behavior or did I hit an odd bug?

Hi, I have os.path.join(path, folder) in my task, and I can run flow with it from python script but not on prefect UI. The error message is: Unexpected error: TypeError('expected str, bytes or os.PathLike object, not NoneType',). Anyone has any thoughts? Thanks in advance.

prefect agent docker start

results in

requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7ff5ba1a23a0>: Failed to establish a new connection: [Errno 111] Connection refused'))