Hi all - if i want to use a python main.py outside of my prefect.py file what is the best way to implement this type of configuration based on usage and deployment? Is there an example somewhere how this ideally should be constructed?

Has anyone seen this kind of error? Everything was fine in testing with local storage, and now that we’ve pushed to prod (which uses s3 storage), we are seeing this when we run our flows almost immediately:

Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n AttributeError("Can\'t get attribute \'create_params_file\' on <module \'our_filename.py\' from \'/our_filename.py\'>")')

The signature of

create_params_file

, which is not a task but a method that can be called looks like:

def create_params_file(base_filename: str, **kwargs) -> str:

How does your team manage flow storage and run types? I have tried local storage, github and docker, I know there are a few others but they seem peculiar. Personally I’m a big fan of docker, but we are debating the functionality/learning-curve trade off. Are local agents running on VMs stable enough for production use? A lot of developers have their hands full learning dbt, python and git - throwing docker at them too might be a little much. Happy to hear any experiences around this.

Hi, I’m trying to store the Results from each Task in a Flow on S3, but I am having some trouble with it. When I set the results object to be an S3Result, nothing ends up being stored in S3. I am able to directly write files with the S3Results object, but the Results from a task are not automatically stored there. I have tried to set checkpointing to True, so I don’t think that is the issue. Here is some example code.

MY_RESULTS = S3Result(bucket='my_bucket_without_s3_prefix',location='my_output_folder')
prefect.config.flows.checkpointing = True
!export PREFECT__FLOWS__CHECKPOINTING=true
with Flow("please work", result=MY_RESULTS) as f:
    t1 = my_task()
state = f.run()
!aws s3 ls <s3://my_bucket_name_witohut_s3_prefix/my_output_folder>  # nothing is here

Is there something obvious, which I am missing?

Hello! I'm testing out orion and have an issue with the client. When I try to run the sample for python client it gives me an error:

line 282, in get_profile_context raise MissingContextError("No profile context found.")

Any suggestions on how to fix this?

hey guys is it possible to use

create_flow_run

to create a persistent scheduled flow run? Not a flow that runs only ones at a scheduled time but a flow that runs at intervals?

Have a question about expected behavior for

map

- With the following sample flow, I was expecting the mapped task

f

to run concurrently with

wait

because there are no dependencies and

LocalDaskExecutor

is being used. However, the behavior I see is that only `wait`'s mapped tasks get executed, so

f

is not executed until all the

wait

tasks return.

from prefect import Flow, task
import time
from prefect.executors import LocalDaskExecutor

@task
def f(x):
    return x*2

@task
def wait(x):
    time.sleep(x)

with Flow("test") as flow:
    a = list(range(4))
    wait.map(a)
    result = f.map(a)

flow.executor = LocalDaskExecutor()

Hi! I have some long running tasks that I cache their output with

target

, however, from time to time I want to re-run these tasks without manually deleting target files. Is there a way to "force" re-run for such tasks?

I currently have a prefect flow that operates on a single row of a pandas dataframe. Is there a straighforward way to map this flow to all of the rows in a pandas dataframe? In other words, can I create a flow and then map it? If I cannot map each row of a dataframe to a flow, is there a straightforward way of nesting different tasks into each other and then mapping that “super” task to a series of inputs?

Hello, I was wondering if there was a way to prevent a scheduled flow that is scheduled to run every 30 mins, to not start if the previous run has not finished. Any ideas how to do that?

I still can’t

pip install prefect[azure]

on my m1 mac 😞

How do you stop a deployment with a scheduled interval?

If I want to run my flows on the ECS fargate, do I setup a DaskExecutor with Fargate with a cluster class? Is it fine if I have an agent local and not using the ECS agent?

Hello, does it make sense to distribute a Prefect task using Dask (e.g. predict a large dataframe with a ML model), in a Prefect Flow that already uses a remote DaskExecutor?

Hello everyone, I see on this issue (https://github.com/PrefectHQ/prefect/issues/1545) that in 2019, there was no easy way to retrieve logs from distributed dask workers. Is there any update that I missed? Have you found a new way to deal with this? Is the creation of a service still the recommended way to achieve log retrieval?

Hi there I'm not able to let orion run with docker run. (more infos in the threads)

Hey! How can i use GraphQL mutation to delete prefect flow run logs by flow run id? an example would be appreciated!

Hello, I'm trying to do some sort of switch scenario where depending on an input parameter it will get data from a DB and get a DataFrame and then go to a common task to transform data and so on. I've attache a description image

Hi, I'm trying to get logs from the GraphQL API but there is a limitation in the query. Is there a way to paginate the query so I can get all of the content?

Hi there! Anyone could explain where I set the 'config', while running SatartFlowRun ? I'm using with

new_flow_context=prefect.context.get('config')

Hey guys 🙂 Got a question on flow scheduling. I’m using the basic IntervalClock for my flows, with all of them being started at pretty much the same time, which worked fine until now. The numbers of flows increased to 30+ and they started stalling, never finishing. It makes sense since we’re querying the same DBs in them and it’s all run on a single EC2. I’d like to adjust the scheduling, so that they’re starting in small groups, few mins apart. My question is - what’s the best way to do that? First thought is CronSchedule, but as the number of flows starts growing into 100's maintaining those individual schedules might be problematic. Is there another way?

Hi folks, since last week we have been encountering an issue with prefect cloud, version locking and heartbeat failures - more details in the thread.

Hi, i have kind of silly question, if i want to run agent, flow and server on same gke cluster can i have local agent instead of kubernetes agent?

I'm looking to run create_flow_runs using threading for a custom module I'm writing and for some reason I'm getting an attribute error. Below is my test code and the error. Would you know how to fix this? Also, I'm assuming this is not a prefect problem and more of a threading problem, but I was hoping someone would be able to help! Thanks in advance

from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
import threading

def thread_flows(flowname):
    print("Running thread for: ",flowname)
    flow_id = create_flow_run.run(flow_name=flowname)
    flow_run = wait_for_flow_run.run(flow_id, stream_logs=True)#

if __name__ == "__main__":
    flow_list = ["FlowA", "FlowB", "FlowC"]

    threads = []
    for flowname in flow_list:
        x = threading.Thread(target = thread_flows, args=(flowname,))
        threads.append(x)
        x.start()

    for thread in threads:
        thread.join()

File "/home/test/.pyenv/versions/3.8.6/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/home/test/Documents/create-flow1.py", line 6, in thread_flows
    self._target(*self._args, **self._kwargs)
  File "/home/test/Documents/create-flow1.py", line 6, in thread_flows
    flow_id = create_flow_run.run(flow_name=flowname)
  File "/home/test/.pyenv/versions/3.8.6/lib/python3.8/site-packages/prefect/tasks/prefect/flow_run.py", line 123, in create_flow_run
    logger = prefect.context.logger
AttributeError: 'Context' object has no attribute 'logger'

Hi guys, i have a question. Is it possible to start another flow, once a task in another flow is completed. For illustration :- Flow A Task 1 > Task 2 > Task 3 Flow B (once Task 2 in Flow A is completed on success, run Flow B) Task 1

Hello guys, I'm using

prefect=2.0a12

, and I would like to schedule my flow like a cronJob. It's possible with decorator?

from prefect import flow, task

@task()
def s1(message):
    print(message)

@flow() # where I put schedule params?
def update_flow() -> None:
    s1("hello prefect")

set the channel topic: Welcome to the Prefect Community! Please use threads if possible so we can archive helpful conversations to: - GitHub: https://github.com/PrefectHQ/prefect/issues?q=label%3A“Prefect+Slack+Community”+ - Discourse: https://discourse.prefect.io/docs

Hi guys, I have flow deployment issue. Agent does not pickup flow registered in different host  Host A 1. I was doing my development on my local Macbook: 2. I created a flow  3. I switched backend to prefect cloud 4. I connected to Prefect cloud with API key. 5. I registered flow to to Prefect Cloud. 6. I started local agent  7. I can see both the flow and agent in Prefect Cloud UI 8. I ran the flow, agent picked up and execute the flow, it worked fine 9. I shutdown the agent Host B 1. I created AWS EC2 instance and deploy my code 2. I connected to Prefect Cloud with same API key  3. I switched backend to Prefect Cloud and started local agent 4. I can see agent in UI 5. However, when I start running Flow I registered earlier, agent does not pickup flow Please advise

Hello everyone, is it possible to add/remove labels from agents through the API with

update_agent_config

? Thanks!

hi a quick question we have created prefect flows in our dev env and using prefect cloud to schedule and run now we need to implement ci/cd to automate deployment to prod env, how can we achieve it we are not using docker for prefect what is the easiest method for ci/cd for prefect