I'm working on using prefect to improve the structure and speed of our data pipeline. We are dealing with lots of data O(10GBs) and I have noticed that when i use

LocalExecutor

the implicit in-memory caching results in the ETL eventually getting oom-killed 😕 I'm using checkpointing and some of the tasks explicitly write their results to disk where they are picked up by subsequent stages so the in-memory caching is not strictly necessary. I have not really found a way how I could tell prefect to not bother with this or at least, not eat all available memory while doing so. Any pointers for how I could solve this problems would be really helpful!

Need an advice on how to use context in cloud Hey there! Glad to be here :-) We use Prefect via Cloud. There some parameters in our flows. I want to store parameters' values in the prefect.context.  Code from the docs works fine locally:

@task(log_stdout=True)
def print_context():
  print('print context from task:', prefect.context.get('db_env'))   
   
DB_ENV = Parameter("database to connect to", default='local')

prefect.context['db_env'] = DB_ENV  # ← doesn't work in cloud

with Flow('testing_context') as flow:
  print_context(upstream_tasks = [DB_ENV])
      
flow.run()

then I register the flow in a docker storage and push it to the cloud. And it doesn't work when I run the flow via cloud ui: it shows None instead of parameter.

Hi has anyone had this error before?

Failed to load and execute Flow's environment: UnpicklingError("invalid load key, '{'.")

Is there a way to prevent the prefect docker flow storage from pip installing the latest and greatest prefect? My build setup/team is aghast. (We prefer to explicitly manage dependencies ourselves.)

Good morning! I’m seeing that for CPU intensive tasks, memory usage is ~4 times (or more) higher when I run a task in a prefect flow context vs not in a prefect flow context. I’m working on putting together a simpler example to post here, but in the meantime was just curious if this is something that is known or has been seen before? We are currently using Prefect version 0.12.0, if relevant. Thanks so much!

Hey prefect team, just wanted to let you know a heads up on installing GCP extras with pip - 🙂 i ran into this issue where it kept saying it needed to be installed as a prefect extra, but realized after i removed the try/except in the _init that_ it's due to this issue: https://github.com/googleapis/google-cloud-python/pull/9826. After upgrading my six package it's all good now, but maybe it's worth rewording the importerror?

Hey there, is this possible? I want to default out another parameters based on a previous parameter. In this example, let’s say

x

is a string and I want to

.lower()

it.

flow = Flow("some flow")
with flow:
    x = Parameter("x")
    y = Parameter("y", default=x.foo())

The use case I will be doing this for is date specification and building dynamic dates based on a main driving parameter

Is there a concept of a task of tasks? I have a few ETL flows that all start with similar steps of

1. List S3 files
2. Filter to S3 files I want to copy
3. Download S3 files
4. Upload to GCP

I’m wondering if I can make a compound task that can parameterize the following steps. Or is the prefect idiom to make a dependent flow? Part of the motivation is laziness, but if we ever change where/how we are starting our data ingestion, it’s be great not to have to change it in multiple places

I think there's something I'm missing about states that I just can't figure out. I want to send a notification of a Task failure but ONLY if the task has exceeded its retries. I can't use is_failed(), because I'd get a notification for each failure, same for is_retrying(). I could look at the run_count and if it's on its very last retry go ahead and send a failure notice, but that's not necessarily true (the final attempt may succeed). I thought maybe is_finished() would only be True on the final execution, but that's not the case. Is there a way to know if a task failed and exceeded its retries? (I'm using a task state handler for this)

Hello Prefect Community! I have a Flow that needs some post-clean-up. The Clean-up tasks is configured to always run, no matter what happened beforehand. Unfortunately that seems to also make my Flow always Succeed. Any way to make the flow fail even if the last Task Succeeded?

Hello Prefect Community! I’ve been testing Prefect and it is amazing 🙂 I am simulating different task flows. I love versioning the flows, but it comes to my mind that, when a flow has a bug (i.e. discovered because new input data raises a new exception) and causes an error in a flow run, I need to deploy a new flow version and restart the failed flow run. In that case, if I restart the failed flow run it uses the previous version. What is the recommended workflow for this use case? To make a manual flow run with the correct parameters? Thank you in advance!

Is there any precedent to creating a custom trigger? For example could I create a trigger that would cause a specific task (not flow) to not be run until a particular time?

Hello Prefect community! Is there a way to make Prefect stop trying to execute a flow? There’s a pod in my Kubernetes cluster that is labelled with a particular flow run ID. It will never succeed because I took down the Docker registry it depends on. If I delete this pod, Prefect will create another one. The button in the Prefect UI to cancel the flow run was greyed out, so I simply deleted the flow run with the

delete_flow_run

GraphQL mutation, thinking that without the flow run's DB entry, Prefect would stop trying to execute it. That removed the flow run from the UI, but if I delete the associated Kubernetes pod, Prefect still recreates it. I found no subcommand in the

prefect

CLI tool that can solve this problem, either. Is there any way to stop Prefect from doing this?

Hi everyone, I ran into the

Could not serialize object of type Success

problem that I've seen a few times on this Slack. From my reading of https://github.com/PrefectHQ/prefect/issues/3625 I suspect it's because I'm using

resouce_manager

classes to hold database connection and HTTP connection pools while I map over everything in my Dask cluster. Is it not possible to share a connection pool across a Dask mapped pipeline (thinking about it now, it would be difficult)? If not how should I go about mapped reading & writing? Thanks.

Hi folks, I am running a graphql mutation to delete a flow but I can still see the flow on the UI even after refreshing - is this because there are older versions of the flow ? I see there is an option to delete all the versions of a flow from the UI, is there an equivalent approach using graphql to do so?

Does Depth-First Execution work if you need to use a

flatten

at some point? Like if you pull a bunch of IDs and it takes the form of a list of lists, and then use those as the basis for a bunch of other calls?

Hi all, new to prefect - quick question. In the past our team used Kylo (a NiFi abstraction) for basic workflows and Airflow for complex workflows. One key requirement is to have a way to define a "template" pipeline. lets say pull from blob storage and dump to staging database. Things like blob storage location and database table names may change and should be tied to a particular run. Kylo did a good job of this; a self service for end users to create flows from NiFi templates - users could create the entire pipeline from UI using the template as base. The Kylo project hasn't been maintend much. Airflow doesn't fit this model as dags have to be developed for each new run instance unless you get into dynamic dags etc. Does Prefect offer something like this in the prefect UI? Let me know if this makes sense or a more detail example is needed.

I’m running Prefect in Kubernetes. What I have is a flow which spawns 15 Kubernetes high CPU intense jobs. In order to do some parallelism, I have in the flow a DaskExecutor configured (6 workers, 1 thread). What I see is the prefect-job which is created by the Kubernetes agent, uses quite some resources (400m CPU, ~850 MiB) . Is there a better alternative to deploy this? Which uses less resources? The alternative is using a static dask cluster, but as multiple flows can be ran at the same time, this doens’t seem like a better alternative. I’m open to some suggestions 🙂

Hi everyone, was wondering if there is any easy way to find the result of all the task inside the flow through GraphQL. The rationale being we trigger the task from Flask and wanted to show the result for each task from Flask UI

Hello all! Does setting+updating a Secret From within a Task is something possible with Prefect Cloud (In Cloud Context) ? I see I can easily do this via the Client/GraphQL, but didn’t manage to find a way to access the task’s current client/GraphQLClient. Any hints on how I could achieve writing to a secret from within a Task?

Hi! I would like increase the retry_delay of a failed task each time the task fails. Is there any way to achieve this?

Hi Prefect experts, I am wondering what is the web server that manages the prefect UI. I do not see that is being part of prefect core server architecture unless the web server is incorporated as one of the component like Apollo etc.

Hello all, I have a more conceptual question today: My company has existing ETL tasks that are all in their own docker containers. I have been successful in having prefect run the container as is. My question, however, is how would we be able to get the most out of prefect. Right now each ETL task lives in its own docker container with its own sort of logging handled by python's logging class. We are wanting to move everything over to being handled by prefect and I am looking for input and advice on how best to utilize the capabilities of Prefect for Flow deployment. I know that flows can be restarted at any stage in which they fail, but how would we be able to utilize this if the task the flow is running on is in a docker container? Are they any docs that I could read, as well as examples? We are wanting to be able to get alerts at any point in the process chain upon failure, i.e. if the API call times out or the task of loading the data into our databases fail. Thank you in advance for your help and advice!

hello all! i am in the process of trying to hack together an ETL flow. i've successfully gotten prefect server running on a remote host in a docker container, and it can run a simple flow with no problem as long as the flow is built into the container's image. what i don't quite follow is how to register another flow without stopping and rebuilding the container--it looks like there used to be a tutorial on using prefect with Docker, but it was removed? https://docs.prefect.io/orchestration/tutorial/docker.html is there an example or tutorial somewhere that explains this? i can't figure out how to run the python script containing the flow registration code in the running prefect environment

Hello everyone ! Sorry If this has already been answered here but my search did not return anything. I have trouble understanding authentication through google cloud to store my results with the help of prefect cloud secrets: Following this part of the documentation, I have setup a secret GCP_CREDENTIALS containing the JSON service account key. From what I understood, this should be enough to automatically authenticate my flows with the right credentials: however, when I run my docker Agent, my task fails with the following traceback:

google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see <https://cloud.google.com/docs/authentication/getting-started|https://cloud.google.com/docs/authentication/getting->started

Is there something I am missing there ? Should I include the google credentials when building my flow's docker image ?

Is it possible to run multiple flows from the same docker image? I’m using a docker agent and have multiple flows with the same set of dependencies. So I wanted all the flows to use the same docker image with different tasks and parameters per flow. When I run register the flows, each register call builds and pushes a docker image for that flow which will start to add up once I have more than a few flows. Is there a way to ask the flow/storage to check if a docker image exists already and only create if it doesn’t exist?

hello, I have a quick question. Could I turn off the schedule of a flow using

Prefect cli

?

Hi everyone, I have a very intriguing question - I have registered my flow on a k8s prefect-server and it's using Dask Gateway as the executor. The flow is actually working just fine, to my surprise... However, I can't see the Dask cluster being created by Dask Gateway in the GCP UI. Therefore, there are no dask workers to execute the job, yet the job is still running fine? Is it even using the Dask executor I specified when I registered the flow?

Is there any way to have a Flow only run a portion of the dag? And if so, is it possible to have a Schedule that only runs a portion of the dag?

Hi everyone, has anyone seen this error when trying to run a kubernetes agent using rbac?

agent | Service token file does not exists. Using out of cluster configuration option.

I’ve confirmed that the service account, role, and. rolebinding all exist on the server. Here’s a screenshot of the pod settings, attaching the rest in a thread Issue was that the deployment did not automount the service user credential. The deployment was created through terraform, which defaulted it to false