I use CLI "prefect register --project ${{ parameters.project }} --label $(dockeragent_label) -p flows/" to register all prefect flow files in "flows" folder. This command runs in an azure devop pipeline. But it seems that the command registers all flows with new version although most of them have no metadata changes. The pipeline run always install the latest prefect version. Any idea why that happens? I though the latest prefect version won't create new version if the metadata is not changed.

Has anyone faced this before? the schedule for this flow is defined as

bot_schedule = Schedule(
    clocks=[
        IntervalClock(
            interval=timedelta(hours=1),
            start_date=datetime(2021, 1, 1),
            labels=[
                ...,
            ],
            parameter_defaults={
                ...,
            },
        ),
    ],
)

At first, we thought it was an UI issue. But then, we've confirmed that the flow wasn't being scheduled at all for this time window. Any ideas?

hi, if I wanted a task to loop or retry by raising a signal, how would I delay the next iteration of the retry? I’m basically create a task that polls something and I’d might want to check that every 10 minutes or so. Previously, I had a for loop/sleep in the task but I figured this signalling would be a better way?

I have a flow that appears to time out during a lengthy sql query with an error message

No heartbeat detected from the flow run; marking the run as failed.

. The SQL query takes ~45 seconds to run on my local machine so I am curious what interval the zombie killer process polls at and if you have any suggestions for debugging this interruption. The flow/associated scripts run fine on my local machine. Thanks.

noticed that I can no longer select States on the Cloud Hooks page (this is a Slack hook specifically), I know it worked semi-recently....any idea what might be up? cc @nicholas

I have a flow that runs a

select * from table

to snowflake using a SnowflakeQuery task. It appears that this task uses

fetchall()

to return this data into memory: https://github.com/PrefectHQ/prefect/blob/5d2732a30563591410cb11fe0f7e7dfe65cc5669/src/prefect/tasks/snowflake/snowflake.py#L186 I expect that this causes performance issues with extremely large queries so I am wondering what Prefect Cloud’s tolerance for this. Ideally I think it would be preferable to lazy load query results and/or allow for query pagination? Perhaps there’s an architecture limitation I’m overlooking? I’d appreciate any insight 🙂

Hello all, myself satyapal from india

intend to use prefect to create workflow in my application

would like to know , whether anyone has tried prefect in this direction?

What is the max timeout for a prefect task? I have a task that will run for 2-5 mins. I couldn’t find anything on the pricing page, google, or discord when I searched for “max timeout.” Is there a better term to search for?

I have registered a flow in prefect cloud and I am trying to run it in Kubernetes (AKS). Right now I can see that the Kubernetes agent submits the flow for execution and the first task in the flow seems to execute fine until it errors with this message. If I run the flow locally all is fine. Has anyone faced the same problem?

Hey folks 👋 We’ve been using Prefect Server for two years now and are really happy with its flexibility! However, keeping Prefect Server up to date is becoming quite expensive (1 Data Engineer + 1 SRE need to spend 1 week every time we want to upgrade). So, we are evaluating the ability to switch to Prefect Cloud. Given that pricing is based on successful task runs, I have few questions: • what’s the best way to get daily task run stats out from Prefect Server? • how does task/flow mapping fit into Prefect Cloud pricing? Regarding the first point, I crafted up a query but I’m not sure it is correct. Any help would be much appreciated! 🙌

with raw_data as
(
	select "start_time"::date as task_run_date,
	       state,
	       count(id) as task_run_per_date_and_state
	from public.task_run
	group by 1,2
)
select state,
       avg(task_run_per_date_and_state)::int as avg_task_runs_per_day
from raw_data
where state = 'Success'
group by 1
order by 1

Hello Prefect Community, I have an issue while installing the latest version of Orion i.e. "Prefect 2.0a13". I see the error while deploying the ray[default]>=1.9. I am on Python 3.7.6 and running this on the Linux platform. Does anyone have a similar issue? Thanks

Hi Community. I have query regarding cloud offering. As I see that first 20000 tasks are free and later are charged based on usages. But I am not able to find other details like system limits (memory, cpu, network, storage and runtime limits)

Hi 🙂 I missclicked in the UI and canceled a scheduled flow, is it possible to "uncancel" it ?

Hey, I have a failed flow run on EKS - i’m trying to restart but for some reason when its spinning up the new pod its requesting CPU that the original run config did not set. From run config on the flow run:

"cpu_request": "4",

On describe pod that is spun up:

Requests:
      cpu:     6
      memory:  16Gi

Why is this happening on the restarted flow run that i’m trying?

I'm working on writing a state handler that executes some logic at the very end of a flow, after all tasks have run. In that logic I end up calling

StartFlowRun(...).run()

- when this executes in the cloud I'm seeing: 🧵

👋 Hi everyone! Just so thankful for you all. Let us tell you, we’re super excited And you should be, too, even very delighted! Until the appearance of ol’ Orion orion Never before had we felt like we were flyin’ Ceremony and excitement for what’s next How can we contain what we think is the best? We are pumped for the future Endless fun & positivity await Eternally grateful for what we’re all creating together 🚀 Keep up the good vibes community, we’ll see you on Monday!

Hi! I am trying to register a flow onto prefect cloud with docker config:

flow.run_configs = DockerRun(image="myrepo/image")
flow.register(project_name)

But it shows up incorrectly as UniversalRun in cloud UI. If I add the runconfig when I define the flow it works perfectly:

with Flow("myflow", DockerRun(image="myrepo/image") as flow:

Any suggestion to what I am doing wrong in the first approach would be appreciated

Is there an upper limit on what version of Prefect is compatible with Prefect Cloud? (obviously not Orion/2.0, looking at jumping to 1.1 right now)

Hi Folks - I have a question about using the return value from a task as input in another task. I have searched around and am not sure if I am missing something. 🧵

Hi, I'm working on implementing opentelemetry traces & metrics for python into my Prefect code. Are there any examples or references for how to structure flows and tasks with them? I was hoping to use state_handlers for referencing the current trace, but doesn't seem to work at all.

Excuse my first question. I want to access a container that is already started in my local environment and run a job. (I don’t want to use DockerRun to create a container for each job, I want to leave the container running.) I would like to know if there is any way to achieve this with PREFECT.

Hi! I am trying to create a directory within a container, something simple that looks like the code snippet below (in the thread). The task is running successfully but failing to create the data directory. I was thinking that it's supposed to create the directory without any issue, any suggestions to what I am missing please? Thank you

Hi, I'm trying to work through the Prefect Orion tutorial for running flows in Kubernetes (https://orion-docs.prefect.io/tutorials/kubernetes-flow-runner/) and I keep running into the same problem. When I run the deployment,

prefect deployment run test-flow/test-deployment

, the UI shows that the flow run is Pending. When I look at the logs in Kubernetes with

kubectl logs -l job-name=voracious-antelope2b89g

, I see a FileNotFound stack trace Thank you for your help!

Hello, Trying to get a performance report based on the information executors documentation. I am getting an error - see in thread

Hey, I’m trying to run a flow with LocalDaskExecutor - this is my code (simplified - I have multiple tasks after those 2)-

CREDENTIALS_SECRET_NAME = 'PREFECT_SERVICE_ACCOUNT_CREDENTIALS'


with Flow('comparable-products-data-extraction-development',
          storage=Docker(registry_url="us-central1-docker.pkg.dev/xxx/",
                         dockerfile="./Dockerfile"), executor=LocalDaskExecutor(scheduler="processes")) as flow:  # , schedule=daily_schedule
    raw_credentials = PrefectSecret(CREDENTIALS_SECRET_NAME)
    # raw_credentials = get_local_credentials()
    google_credentials = parse_credentials(credentials=raw_credentials)

And here is the parse_credentials task -

from prefect import task
from google.oauth2 import service_account


@task()
def parse_credentials(credentials):
    return service_account.Credentials.from_service_account_info(credentials)

This works fine in multiple flows when not using Dask executor - and it also runs without any problem locally when using Dask…. When running this in Prefect cloud - (PREFECT_SERVICE_ACCOUNT_CREDENTIALS is saved as a json secret, and it works in other flows) I get this Dask error - (see at the my first comment) Any solution for this? what am I missing? Thanks

Heya Prefectionists, I’ve built something of a dumb s3 sensor, but it appears to get stuck whenever it attempts to retry. Any suggestions or ideas as to why it could hang on retry? I’ll include a snippet of the code in the comments, all the upstream tasks have result types in PrefectResult().

Hi all, I'm using prefect core, and when I

run prefect server start

I get the following error

Pulling postgres ... error

Pulling hasura   ... error

Pulling graphql  ... error

Pulling apollo   ... error

Pulling towel    ... error

Pulling ui       ...

Hello, Is this the correct way to ensure a task only runs once a week, in a flow scheduled to run daily?

@task(result=LocalResult(dir=f'{os.getcwd()}/storage/pre_delinquency_models',
                         serializer=H2OModelSerializer()),
      checkpoint=True, target='{date:%Y}-Week {date:%U}', log_stdout=True)
def train_new_model(data: pd.DataFrame):

I have it running on prefect cloud with a run_config that looks like below:

run_config = LocalRun(env={'PREFECT__FLOWS__CHECKPOINTING':os.environ.get('PREFECT__FLOWS__CHECKPOINTING', 'true')})

However, looking at the results folder I see that the task result file is created a new with each daily run . What am I missing?