Good morning! Since the upgrade to

2.0b8

we have experienced 2 things: • performance of single flows became around 2x slower (we are using the Sequential task runner) and have a lot of database operations in our tasks • Some Flows get interrupted by a 403 forbidden error and do not pick up from where they were left. Is there a way of increasing the lifetime of the token?

Apologies in advance for quite a vague question 🙂 I've historically custom rolled data pipelines (janky shell/python scripts, slightly nicer but terrible observability with Camel). Is there much usage, or nice ways for dealing with time series and spatial data in Prefect? I can see for example being able to look at success/fail rates of the run history, I'm wondering how this could work when sometimes you only "care" about a failure for a short amount of time and that's subject to when it last failed. Use case: satellite images are processed for various areas - there could be one job for each tile of interest. Periodically each tile is checked for new data and when there is some, a number of steps to load and process are performed. Sometimes there isn't data, so the first step fails. This can be because it isn't available yet, or it won't ever be available. Failed might be fine because there's nothing we can do anyway (there's just no data). However if the same one is failing for say a couple of weeks and there's no "recent" data then it becomes bad. Sometimes bad things happen and everything fails. Maybe an API key has been revoked or a service is down. Failed is bad. For my uses I ended up tracking these and making a dashboard showing e.g. each day along an x axis and each tile along a y axis to be able to spot gaps which hopefully might go from red to green after retries. Is something like that possible via Prefect? To pull out and create custom dashboards like that or via API calls etc? This is purely an "out of interest" rather than an immediate need, observability and even being able to categorise failures was a huge problem so I'm just wondering if this is something catered for, so if there are solutions that would be great to here, but if not "nope, you're still on your own" is fine 😄

[SOLVED] Hello, a newbie question I am getting started with prefect, when I run `

prefect server start

I get this error:

OSError: [Errno 8] Exec format error: 'docker-compose'

I suspect it might be a

docker-compose

version issue, since the last version is called by running

docker compose

(without

-

) Should I downgrade? (Cuurently Docker Compose version v2.6.0)

is there a specific reason https://github.com/PrefectHQ/prefect/pull/5729 wasn't merged yet? It only changes a few lines of code and helps anyone who has only access to custom S3 endpoints.

hello, in Prefect 2.0 I’ve defined a new custom block as described in https://orion-docs.prefect.io/concepts/blocks/#creating-blocks. Once I’ve defined the model, is there a way to upload the definition of the block in Prefect cloud without having to create a block instance from code and the save it? I mean, I want to be able to see in the UI the new custom block definition without having to define a dummy instance of it

I have a flow that runs every month. I’d like the flow run name to always include the name of the month like “My Flow - July”. What’s the best way to do this?

hello!! I noticed that Prefect UI will print out the ECS task id in the log output while the task starts running, I was wondering is there any way to directly retrieve it?

is there an open source ticket to track when the

checkpoint = False

flag will be introduced into

orion

/ prefect 2.0?

hi! it's failing to create a project from cli and from code

prefect.exceptions.ClientError: [{'path': ['create_project'], 'message': 'Uniqueness violation.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

From inside a task, how can I get the start time of the current flow run?

i have a component that is extremely costly to instantiate that needs to be used in multiple tasks. is there a way to distribute an instance of this as a singleton so that i do not need to reinitialize the costly processor each time the task is run? i'll provide a basic model of the python code demonstrating my dilemma

@Anna Geller - We are seeing one issue with our flow today. We have scheduled our flow every 5 minutes. We use KubernetesRun as our run config and S3 as our storage. We have scheduled the flow from the Prefect Cloud. The issue that we are seeing is that if the flow run’s run time exceeds 5 minutes, a second flow run of the flow runs in parallel. Today because of upstream long running jobs, our first flow run was running for 15 minutes and in the meanwhile there were 3 flow runs of the same flow that were running. Is there any way to ensure that a new flow run runs only when the previous flow run completes?

Hi community, how is the best practice for implementing parallel dbt jobs using Prefect 1.0? Basically, I need to send different client_id to dbt jobs and trigger dbt run for each client_id (pseudo code as following)

with Flow as flow:
  for i in id_list:
    dbt_run_function(i)

(I’m wondering if a simple for loop would achieve parallelism?)

Hi community! I am trying to use a monorepo where each folder represents a flow, currently I am using ECS Agent to run the flow on AWS Fargate. I am a little bit lost on the errors I am currently having. I am trying to build a docker image for each folder since it has a flow.py file and multiple others like extract.py, load.py, etc ... that are imported from flow.py and then registering the flow, is this process correct and if not which would be a better solution ?

Hi All - is there a best practice when it comes to parallelization within tasks (I'm using local Dask cluster for reference)? My initial flow used task mapping but there were tens of thousands of mapped items which quickly burn through the free tier limit. My current thought is to just test basic parallelization (e.g., Python's multiprocessing) within a task but I worry that it will interfere with Dask's use of resources Thanks in advance for any suggestions

Hi all! I’m doing some testing with the DaskExecutor and spinning up an ephemeral Dask cluster on ECS Fargate. Although it seems to work, I ran into the same issue that was described here. I understand that the clean spin-down issue might not be super high priority since the workers seem to execute what is assigned to them but I did have a question about logging. Specifically, my logs look more like this (Beginning Flow Run, then Flow Run SUCCESS, then the RuntimeError) with none of the intermediary logging being captured from the

list_sum

function in the flow. Is there a configuration I might be missing that enables this, or has there been any progress in getting the output of DaskExecutor logs into Prefect Cloud?

I don’t see the paused or skipped states in the Orion docs. This seems like a loss in functionality. What is replacing these?

Is there an intention/movement on putting a Terraform module for Prefect in place? Our Prefect Automations have been manually created and I’d like to have this in source control/managed through a CICD pipeline. Last resort would be to use GraphQL but Terraform would play better with our current stack.

Hello! Following the lastest annoncement about the pip install command, I need some clarifiction 🧐 I'm still using prefect 1.0, not ready yet for migration, do I just need to be carefull about the prefect version installed in my environment to register the flow, or is there other modifications to do ? thanks!

In Prefect 2.0, I’m trying to create a Deployment which takes arguments from a configuration stored in Prefect Cloud as a custom block. Here is the sample code:

config = CustomConfig.load('my_block')

Deployment(
    name="My deployment",
    flow=my_flow,
    parameters={
        "password": config.password
    }
)

when I run this command from CLI

prefect deployment create my_deployment.py

I get this error

AttributeError: 'coroutine' object has no attribute 'password'

Failed to load deployments from 'my_deployment.py'
sys:1: RuntimeWarning: coroutine 'Block.load' was never awaited

How can I do that?

Regarding Orion: The concept docs contain the following example for RemoteFileSystem:

fs = RemoteFileSystem(basepath="<s3://my-bucket/folder/>")

fs.write_path("foo", b"hello")

fs.save("dev-s3")

What das fs.write_path do? Store actual content on the fs? Also, is the secret key stored in the prefect database if you pass them to the settings parameter of RemoteFileSystem and then call fs.save? If so, how is ensured only I can read it and not e.g. a admin at prefect?

Hey, I'm using the latest prefect 2.0 b9 and when running "prefect storage ls" all my created blocks are shown - not only storage blocks. Is this expected? (It's actually not a blocking problem, just wanted to report, if not already known).

Im trying to use local dask executor and add task dependency from the previous task like below (Using prefect 1) but running this gives me and error saying

add_json_index() missing 1 required positional argument: 'output'

output = filter_empty_responses(data)

    # Parsing Jsons
    indexed_output = add_json_index(output, upstream_tasks=[filter_empty_responses])

Hello, I am trying to use the vertex agent unfortunately I keep getting

400 List of found errors:	1.Field: job_spec.worker_pool_specs[0].container_spec.env[12].value; Message: Required field is not set.	2.Field: job_spec.worker_pool_specs[0].container_spec.env[3].value; Message: Required field is not set.	3.Field: job_spec.worker_pool_specs[0].container_spec.env[4].value; Message: Required field is not set.	 [field_violations {
  field: "job_spec.worker_pool_specs[0].container_spec.env[12].value"
  description: "Required field is not set."
}
field_violations {
  field: "job_spec.worker_pool_specs[0].container_spec.env[3].value"
  description: "Required field is not set."
}
field_violations {
  field: "job_spec.worker_pool_specs[0].container_spec.env[4].value"
  description: "Required field is not set."
}
]

Hi community, if we do not use flow of flows framework, is there any downside? Basically from my understanding flow of flows is to use one single flow (orchestrator) to manage other flows (dependency, schedule etc.) Instead of defining those relationship in each flow's script?

In Orion, what goes into the serialized representation of a flow when it’s registered? How can the DAG of a flow be visualized before the flow is actually executed?

i have a flow in which there are numerous subflows and the dependencies of subflow B are depended on subflow A . is there a way to use subflowB.set_upstream(subflow_A) in prefect 2.0?

Hi, I am running into issues with

LocalDaskExecutor

and I cannot understand the problem. I am trying to run parallel calls for Rest API endpoints. The problem is that the Dask executor will run some calls twice for no apparent reason. Here is the logic I have Then I can see that some Dask tasks run on the same endpoint while when I log all the endpoints, there are all unique... I will post the logic as a snippet in the thread 🧵

Hi Community, could you please help me to understand where should I place GITHUB_ACCESS_TOKEN? I have running agent with kube and defined there environment variable

PREFECT__CONTEXT__SECRETS_GITHUB_ACCESS_TOKEN

Registered flow with GitHubStorage

flow.storage = GitHub(
    repo="company/prefect-workflows",
    path="flows/benchmarks/flow_import_index_data.py",
    ref="master",
    access_token_secret="GITHUB_ACCESS_TOKEN",
)

but when I run flow from cloud.prefect.com I am getting error

Failed to load and execute flow run: ValueError('Local Secret "GITHUB_ACCESS_TOKEN" was not found.')

I have been running into an issue with Prefect cloud 1.0 where my flow and task concurrency slots are being used up while the UI reports that no flows are running. There are also flows that have been "cancelled" using the UI and still show up as greyed out boxes with "cancelling..." written for many days. Has anyone run into this issue before?