Hi all small question, Is is possible to name the runs themselves? I'm using Prefect Core and it's making up the names for the run itself. I would prefer to make own myself with timestamp_variable so that I can use the same flow for multiple configs (yaml files).

Hi all! I am having problems with running Prefect through a conda environment. I am creating the env from a .yml file, and packaging this in a docker container. However, when I build the docker container, and run a flow that imports my dependencies, the modules are not found by Prefect (ModuleNotFoundError). All works well when I just run pip install in the docker file, its when I create a conda env that things go wrong. Please find attached a screenshot of my docker and environment file. Thank you in advance!

Hi! I would be very interested in this as well. I already posted a corresponding discussion topic:

Sorry I'm kind of a Docker NOOB. I opted for a miniconda container and installed prefect afterwards. My code is not installed via pip but simply lies in the PYTHONPATH.

Hi everyone — I have an ECS Service with 1 container running a Prefect ECS Agent, however it seems like the memory consumption of this process steadily increases until eventually dies of OOM and then restarts. Does anyone know why this happens and how to fix it? The container entrypoint is simply:

import prefect
from prefect.agent.ecs.agent import ECSAgent

agent = ECSAgent(
    region_name="eu-west-1",
    cluster="knightsbridge",
    labels=["ecs"],
    launch_type="EC2",
    agent_address="<http://0.0.0.0:8009>",
)

agent.start()

Hi all, is it possible to configure the Lazarus to trigger fail after N > 3 of scheduling attempts? or wait more time between reschedules? i have flows that are stuck on a queue due to compute limitation and the lazarus kills them before running

What is the best practise for splitting a large workflow with many upstream and downstream datasets into smaller flows? We are starting to set up our first couple of flows to manage the pipelines for our data warehouse. Like in most data infrastructures we will probably have several upstream datasets, which will feed into the pipelines for datasets further downstream, which in turn feed into more datasets even further downstream and so on. My question is how to best organise a large workflow like this? You could create one massive flow, but this wouldn’t be very nice to manage. So what is the best way to split this up in prefect? Some ideas: • Import a task from the upstream flow into the downstream flow. I don’t think this has the intended effect. You end up re-registering the upstream flow every time you do the import, and I think it ends up just duplicating the imported task, rather than making the downstream flow actually wait for the upstream flow. • Create a waiter task that succeeds once the upstream data has landed.  • Send some kind of event from the upstream flow that triggers the downstream flow. (Not completely sure how I’d do that) • Create a Parent-flow that triggers a bunch of sub-flows Is there a “correct” way to do this? What are people’s experiences? Sorry if I missed an obvious piece of documentation or discussion somewhere.

Hi everyone. Are there any docs that explain how one might configure Prefect Cloud to support different environments (dev, stage, prod)? For example, do we separate using different tenants/projects/agent-labels? How should we manage different sets of secrets and access to those? I'd appreciate any suggestions. We are using Kubernetes for execution. Thanks!

Hi, I am getting this error when trying to start my local agent. I've already authenticated using my token in the cli.

RuntimeError: Error while contacting API at <https://api.prefect.io>

Hi everyone — I have a flow that runs correctly with “prefect run”, however if I try to run via local agent then I get a weird error and the flow hangs in the “submitted” state

Hi all, I'm new to Prefect and trying to set up a flow with Azure Blob Storage. How can I convert the downloaded file to a pandas dataframe? I don't think I'm properly understanding and using the Class. Thanks!

Hi all, question on deploying flows with multiple agents. If I have version 1 of my flow deployed to an agent with label ‘prod’ and I want to test out a new feature on an agent with label ‘dev’. Will this new flow deployment of version 2 on agent dev have an effect on my version 1 flow already running on agent with label ‘prod’?

Hi. When using the new tasks

create_flow_run

 and

get_task_run_result

, how can I assign a different name to different instances so that I can see which task instance is running in the UI without having to look at the logs?

Hi - Is there a way to delete an Agent from the Cloud UI? Some time ago I remember being able do it. I also tried to delete an agent using graphql ... Is this right?

mutation{

delete_agent(input: {agent_id:"<id>"}){success}

}

Thanks in advance

How do we store secrets using Prefect Cloud? I can't find anything in the UI.

Hi, how do I run multiple tasks? I have a list of ids from my back end that I need to rerun the tasks in prefect, but seems like i need to do it one by one, is there a way to pass that list and it will create all the tasks and run them?

How do I set the location of the result upon writing? Below code doesn't work, does not generate the

df.parquet

file:

parquet_result = LocalResult(dir="./test_prefect", serializer = PandasSerializer("parquet")

@task
def test_task(df1, df2):
    parquet_result.write(df1, location = "df1.parquet", **context)
    parquet_result.write(df2, location = "df2.parquet", **context)

Currently I have to set the

location

attribute at the time of instantiating the

LocalResult

object. The code below works

parquet_result_partial = partial(LocalResult, dir="./test_prefect", serializer = PandasSerializer("parquet"))

@task
def test_task(df1, df2):
    parquet_result_partial(location = "df1.parquet").write(df1, **context)
    parquet_result_partial(location = "df1.parquet").write(df2, **context)

So it seems the

location

kwargs to

Result.write

does not do anything. Is this by design? Or am I missing something

Hi there! I have a shared task that I would like to run from a bunch of different flows. The only differences between the flows are the flow names, the schedules, plus a single parameter. Is this a reasonable way to accomplish this?

from prefect import Flow, Parameter, task
from prefect.schedules import CronSchedule


@task
def my_task(my_parameter):
    print(f"my_parameter value is {my_parameter}")


with Flow("my-flow-1", schedule=CronSchedule("0 * * * *")) as flow1:
    param = Parameter("my_parameter", default="flow-1-parameter-value")
    my_task(param)

with Flow("my-flow-2", schedule=CronSchedule("1 * * * *")) as flow2:
    param = Parameter("my_parameter", default="flow-2-parameter-value")
    my_task(param)

Hi, I am getting this error when trying to set up aws with ecs

with Flow("user_brand_campaigns", storage=STORAGE, config=RUN_CONFIG) as user_profile_flow:
TypeError: __init__() got an unexpected keyword argument 'config'

Hey everyone, We're using a test setup very similar to the one detailed here. The only difference is that we're not using any tags for the local agent, and most flows use LocalDaskExecutor, and used compose scale to get a total of 10 agents.  We have observer a few of issues and would like to clear them before we can decide to implement prefect. We noticed several tasks are logging multiple times, in tandem with the number of running agents. When we used 3, we would get 3 log messages for each task status, or custom messages too, for that matter. When we scaled to 10 agents, we get 10 messages, as shown in the print. Not sure if related, but I noticed too that the docker logs seemed to show that a certain agent that only picked up a single flow  nevertheless seemed to deploy every flow that had been scheduled for that 20h scheduled time. So, it looks like every agent is trying to pickup every flow? I have also noticed a few cases where the server seemed to be showing statuses for tasks that would break their dependency chain (example below), showing earlier tasks being killed by zombie killer, but later ones marked as success. Also, many long running tasks show as failure, but generate their proper outputs. Maybe my basic setup can't handle all the requests being thrown at the API? For long running tasks, it looks like the only way would be to turn off zombie killer? From the source code it appears the 10 minute heartbeat delay is hard-coded, so if I have a task expected to run for 20m it'll be forced to fail every time? I appreciate if anyone can clear these up and guide me towards a proper configuration Thanks!

Hello! prefect newbie here! our company is considering implementing prefect to our existing data pipeline process and was wondering Is it possible to configure prefect with multiple ec2 instances? we have different servers with ml training job, and batch job and would like to use one prefect to manage all tasks in different instances I saw the concepts of ‘agent’ so in this case would it be producing agent in each server and connecting them to master node? Thanks in advance!

Hey folks, I have a question concerning the MySQLFetch task, since it appears I can't supply a file to the query parameter e.g. task = MySQLFetch(query="./data/test-query.sql", ... Is this on the roadmap since otherwise I will have to load each file and pass into the task as a string manually ... thanks for any pointers!

If I use Prefect Cloud to schedule flows and have a Local agent for the execution. Do I need use .run() while registering the flow and keep this terminal open?

Hi, I'm trying to connect to SQL Server but am running into some issues. This is a condensed version of my code ^^. Any ideas why I get this TypeError?

Traceback (most recent call last):
  File "C:\Users\***\code\pipeline\src\flows_combined_entries.py", line 76, in <module>
    sqlserver.run(query='select * from julitest.entries where id = 50299',
  File "C:\Users\***\code\pipeline\venv\lib\site-packages\prefect\utilities\tasks.py", line 441, in method
    return run_method(self, *args, **kwargs)
  File "C:\Users\***\code\pipeline\venv\lib\site-packages\prefect\tasks\sql_server\sql_server.py", line 90, in run
    executed = cursor.execute(query=query, vars=data)
TypeError: execute() takes no keyword arguments

Has anyone ever seen this error code before? One of my scheduled flows failed early this AM – after the flow was successfully downloaded, it tried to start up its first task, and then ~10 minutes later failed with the error:

prefect.utilities.exceptions.ClientError: [{'path': ['flow_run_by_pk'], 'message': 'request to <http://hasura:3000/v1alpha1/graphql> failed, reason: read ECONNRESET', 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'request to <http://hasura:3000/v1alpha1/graphql> failed, reason: read ECONNRESET', 'type': 'system', 'errno': 'ECONNRESET', 'code': 'ECONNRESET'}}}]

I haven't run into this one before...this is on Cloud

Hi everyone. What is the recommended way to "forward" the parent flow's labels (at run time) to a child flow?

Hello everyone, what's the default directory for the configfile in windows? i can't find it , when runing prefect config i get that it's pulling its config from C:\Users\username\.prefect ,this directory does not exit on my machine , thx

PS C:\Users\username> prefect config
{"debug": false, "home_dir": "C:\\Users\\username/.prefect", "backend": "server", "server": {"host": "<http://localhost>", "port": 4200, "host_port": 4200, "endpoint": "<http://localhost:4200>", "database": {"host": "localhost", "port": 5432, "host_port": 5432, "name": "prefect_server", "username": "prefect", "password": "test-password", "connection_url": "<postgresql://prefect>:test-password@localhost:5432/prefect_server", "volume_path": "C:\\Users\\Mohamed Hajji/.prefect/pg_data"}, "graphql": {"host": "0.0.0.0", "port": 4201, "host_port": 4201, "debug": false, "path": "/graphql/"}, "hasura": {"host": "localhost", "port": 3000, "host_port": 3000, "admin_secret": "", "claims_namespace": "hasura-claims", "graphql_url": "<http://localhost:3000/v1alpha1/graphql>", "ws_url": "<ws://localhost:3000/v1alpha1/graphql>", "execute_retry_seconds": 10}, "ui": {"host": "<http://localhost>", "port": 8080, "host_port": 8080, "endpoint": "<http://localhost:8080>", "apollo_url": "<http://localhost:4200/graphql>"}, "telemetry": {"enabled": true}}, "cloud": {"api": "<http://localhost:4200>", "endpoint": "<https://api.prefect.io>", "graphql": "<http://localhost:4200/graphql>", "use_local_secrets": true, "heartbeat_interval": 30.0, "check_cancellation_interval": 15.0, "diagnostics": false, "request_timeout": 15, "send_flow_run_logs": true, "logging_heartbeat": 5, "queue_interval": 30.0, "api_key": "", "tenant_id": "", "agent": {"name": "agent", "labels": [], "level": "INFO", "auth_token": "", "agent_address": "", "resource_manager": {"loop_interval": 60}}}, "logging": {"level": "INFO", "format": "[%(asctime)s] %(levelname)s - %(name)s | %(message)s", "log_attributes": [], "datefmt": "%Y-%m-%d %H:%M:%S%z", "extra_loggers": []}, "flows": {"eager_edge_validation": false, "run_on_schedule": true, "checkpointing": false, "defaults": {"storage": {"add_default_labels": true, "default_class": "prefect.storage.Local"}}}, "tasks": {"defaults": {"max_retries": 0, "retry_delay": null, "timeout": null}}, "engine": {"executor": {"default_class": "prefect.executors.LocalExecutor", "dask": {"address": "", "cluster_class": "distributed.deploy.local.LocalCluster"}}, "flow_runner": {"default_class": "prefect.engine.flow_runner.FlowRunner"}, "task_runner": {"default_class": "prefect.engine.task_runner.TaskRunner"}}}