Hey guys, @Dana Merrick and I are using

LocalDaskExecutor

in combination with

RunNamespacedJob

to launch a bunch of trivially parallelizable child processes on Kubernetes. However, it looks like only 2 child jobs are getting run at a time. Is there a way to increase the parallelism // are we doing this right? 😄

Hi! What would be “the right way” in Prefect to do what Luigi refers to as ExternalTask (e.g. a file in S3 that gets created by some external process, and the rest of the pipeline can not proceed without it) ?

Hello all, I just wanted to rule out a possibility with my autoscaling issues I'm facing right now. If we're paying for unlimited concurrency (pay per run), is there anything else in Prefect that would prevent the scheduling of all the pods at once?

Is there an equivalent code for command ?

prefect backend server (cloud)

Right now I am doing

os.system("prefect backend cloud")

, not sure whether there is a more elegant way

I know it's a problem with my code, but can someone give me some tips on a solution ? (I've Googled, but I'm only getting documentation) Error is : Unexpected error: AttributeError("'FunctionTask' object has no attribute load. Did you call this object within a function that should have beendecorated with @prefect.task?") Yes, no space between been and decorated. I have made a few changes, but I didn't think it was anything major enough to cause this error 😞 I've tried back tracking but still can't find the glitch

Hi everyone, I want to ask something about RunConfig and Executor. Im using KubernetesAgent on GKE. My KubernetesRun config was:

KubernetesRun(
    labels=["agent:k8s-dev"],
    cpu_limit=8,
    memory_limit="12Gi",
    cpu_request=0.5,
    memory_request="512Mi",
)

And my executor was:

DaskExecutor(adapt_kwargs={"maximum": 16, "minimum": 1})

When I start a flowrun with these config, prefect agent create a new pod with this config: (image attached) The problem is: The dask worker always killed first before the GKE pod going for upscale, and prefect logging says error KilledWorker Any idea how to perfectly adjust between DaskExecutor and KubernetesRun?

how can i get the state of a task from a task object?

Hi Prefect, I'm running into the following warning when running my flow:

/venv/lib/python3.9/site-packages/prefect/engine/task_runner.py:865: UserWarning: This task is running in a daemonic subprocess; consequently Prefect can only enforce a soft timeout limit, i.e., if your Task reaches its timeout limit it will enter a TimedOut state but continue running in the background.

I'm running a scheduled flow via a Docker Agent with DaskExecutors (multiple processes due to CPU intensive nature of the task). Some of the tasks in the flow can hang and thus I've added a timeout on the task. This warning scares me because, if I interpret it correctly, it might mean that certain processes might hang indefinitely. What is the best way to avoid this and still benefit from running multiple CPU intensive tasks in parallel with Prefect? (I'm running Prefect 0.14.19 and Dask 2021.5.0).

Good morning everyone 🙂 I need some help with prefect. I'm trying to set up a use case where a flow runs subflow. But I got strange behaviors and I don't know why prefect works like this. I put a small example below. My goal is to launch every night a flow that iterates over a list and for each element runs a new flow. Each subflow is configured with the value of this list (it's an url). At the beggining, I was calling StartFlowRun without the idempotency_key. And subflows were called only for the first item. After this, I have used the idempotency_key configured with the item value. I thought It was working but I got strange bevaviors while scheduling my_flow. The flow has worked the first night, but for the others days subflows were not called (like if they were stuck). Then I put a random key for the idempotency_key. And this time everything works correctly for me. Is this normal ? I don't know if it is a good idea to use prefect like this because the GUI will be flooded with new flows. Does prefect rotate the postgres database to remove old flows ?

from prefect import Flow, task
from prefect.tasks.prefect import StartFlowRun
from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock
from uuid import uuid4

@task
def run_subflow(item):
        StartFlowRun(flow_name="my_sub_flow", project_name="test", wait=True, parameters={"item":item}).run(idempotency_key=str(uuid4()))
    

with Flow("my_flow", schedule=Schedule(clocks=[CronClock("0 0 * * *")]),) as flow:
    run_subflow.map(item=["a", "b", "c"])
flow.run()

Hi all Can someone elaborate on this error “Cannot provide

task_definition_arn

when using

Docker

storage”? I have a flow that I want to register with docker (ECR) but I dont want to generate_task_definition every deploy (because besides the docker itself nothing changed) Why cant I pass task arn when using docker storage? Thanks

Hi all Starting from 0.14.17 (also in 0.14.19) the gitlab storage does not work anymore:

Failed to load and execute Flow's environment: GitlabGetError('404 Project Not Found')

We use it with a own hosted gitlab environment. Beside of the new based docker image, we changed nothing:

prefecthq/prefect:0.14.12-python3.8 -> prefecthq/prefect:0.14.17-python3.8

prefecthq/prefect:0.14.12-python3.8 -> prefecthq/prefect:0.14.19-python3.8

Till 0.14.16 it's working:

prefecthq/prefect:0.14.12-python3.8 -> prefecthq/prefect:0.14.16-python3.8

The new Git storage is not able to handle the company's certificate (this was my first try): git storage: ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1131) Are there any problems known? We use it like this in the flow configuration:

, storage=GitLab(
    host="<https://gitlab.xxx.ch>"
    , repo="group/repo"
    , path="test_flow.py"
    , ref="feature"
    , secrets=["GITLAB_ACCESS_TOKEN"]
)

Hi Prefect, It seems that a Tasks

on_failure

handler is called even when there is retry logic added to the Task, and the task still hasn't exhausted its

max_retries

. (I think I just had a Task on which the

on_failure

handler was called, even though it succeeded in a retry.) Is there a way to add a failure handler that is called only after the task has exhausted all its retries and has truly failed?

Hi Prefect 👋 Is there a way to not keep any data into the

flows

folder (my folder is about 200GB of size. from just 2 days worth of flows)? Ideally I would like to delete the flows data once a flow is completed with 100% success.

Hi folks. We’re still trying to move to github storage from Docker, but have run into a new snag. It seems like our toml config files are no longer being picked up. We’ve always configured them at registration time as

PREFECT__USER_CONFIG_PATH=prod.toml

but

prefect.config

no longer seems to have the values from the toml file. Any idea how we can work around this?

Hey y’all, What’s the best way to batch items after running a map. So, as an example, let’s say I’m processing rows from CSV in parallel and writing them into a DB. Writing each in a separate map is quite in efficient. I can reduce them with a into batches of 10000K items. But then the reducer task becomes a single source of bottle neck. Is there any other efficient approaches to batch to items together?

hey yall! I would like to connect with any leaders that have any experience with bootstrapping a Data Governance focused team. We are looking to start up an initiative team around this and would love to hear any experiences that might be out there. Job titles and description nuances compared to standard DE or AE or DA. Let me know if anyone has time to chat and I can DM

Got a question about the GQL query tool: How difficult would it be to query the execution time of all task instances in a given flow with a given identifier?

Have a question about Prefect Context. I Context document under Core Concepts / Execution as well as API Latest / prefect.utilities Context. Both seem to say that users can store items in the context, being careful to not step on any Prefect items, of course, and then retrieve them later. I am attempting to do that but running into trouble. I have created a simple task which receives a "label" to store a time value under. For example, "Start_Flow" would be used to store a time.time() value into the context. That part seems to be working correctly in the task as I can store and then print the stored value. However, in a subsequent task, when I retrieve the "Start_Flow" item from the context, I get a None value. Is there something that I am missing about Context? I am using code similar to the following:

To store:

prefect.context[label] = time.time()

To retrieve:

time_val = prefect.context.get(label)

Hi all, I've got a question about the workflow for upgrading the version of Prefect you're working with. I'm doing an upgrade right now where my plan is to first upgrade the version that my prefect agents are running, and from there I'm going to update the version of prefect in the RunConfig images that my flows are running with; my main question is around what will happen if I update the RunConfig images to have a new version of prefect (in my case, 0.14.17) installed, but I don't yet re-register the actual flows themselves (so they'll still show as having been registered with the old version of prefect I've been using). Is there a huge cause for concern here? When you upgrade the version of prefect that you're "running", do you also need to make sure you re-register all of your flows?

Hi, does anybody know if it is possible to register Worker Plugins when connecting to a Dask Cluster? (as

client.register_worker_plugin

allows)

Hi everybody! I was just wondering why my task not running in the correct order when I'm using

case

.

Hi there! I’m running into an issue where an exception is raised, my code still continues. This however doesn’t really agree with Prefect (understandably). How I notice this? I can see the exception being logged, and Prefect already noticed that the task failed. However, the task still continues, and I see more logs. Check the thread for the details ➡️

Hey ! Quick question regarding automations in Prefect Cloud. It seems the automation isn't triggered if the flow fails to start, for instance if it couldn't be found in the registered storage. Is it expected ? Is that kind of automation part of the agent automation, and not the flow run ?

Hi there again, got different error. I’m running dbt through prefect with custom ShellTask (thus not the existing DbtShellTask). I tried couple commands and it works fine. However now Im running

dbt test

which generates loads of logging output, and it seems something starts to break then:

RuntimeError: reentrant call inside <_io.BufferedWriter name='logs/dbt.log'>

Hi 👋 I was using prefect on localhost:8080 locally (not clould) and for sorme reason, now it started to re-direct me to cloud every time I go to

localhost:8080

. and I can’t go back to my local prefect… why is that and how can I fix this? 🤔

HELP - How do I get prefect secret to work on prefect that is deployed on kubernetes via helm? as you can see below it seems Secrets are grayed out 😞 along with a bunch of other things

Hi all. I have a set of inputs that I am going to map to a series of tasks that dynamically touch databases. Consequently, I don't want to directly pass database connections (to avoid any nonsense in a distributed environment). If I have:

@task
def generate_inputs_from_params(args):
    #....

@task
def create_db_conn(args):
    # ....

@task 
def do_db_work(args):
    # ....

@task
def do_some_other_work(args):
    # ...

with Flow("example") as f:
    # Parameter tasks
    conn_inputs, db_work_inputs, other_work_inputs = generate_inputs_from_params(args) # from param tasks
    conn_map = create_db_conn.map(conn_inputs)
    res_map = do_db_work.map(conn_map, db_work_inputs)
    res2_map = do_some_other_work(res_map, other_work_inputs)
    # some reduce func if necessary

I have two questions about this: 1. Is that flow constructed properly - I'm ultimately after something like:

inputs = generate_inputs_from_params(args) # from param tasks
for (conn_input, db_work_input, other_work_input) in inputs:
    conn = create_db_conn(conn_input)
    res = do_db_work(conn, db_work_input)
    res2 = do_some_other_work(res2, other_work_input)

2. When mapping over credentials dynamically, would I inject `Secret`s into the

conn_inputs

or would I resolve the proper

Secret

"within the for loop" (i.e. extending the map chain to include an additional layer that resolves `Secret`s)? My understanding of docs is that if I do the former, the secret data would be exposed in

conn_inputs

, which in a distributed environment means that plaintext credentials could be making network hops, or in any environment would be persisted as part of

Result

instances. I'd like to make sure I'm understanding this correctly.

Is there a way to have a task take several tasks as argument (in a list), have it run even if one of the argument fails, and have the list passed as parameter contain the exception(s) + the results ? I'm asking because I'd like to have a Slack report based on the results of several tasks, and either I'm adding an handler for each task but then I have three slack messages, or I add it at the Flow level but then I don't have (or at least I don't know how) access to the state and results of each task.

Hi Prefect community, for anyone attending this week's Dask Summit online conference, I'll be giving one of the keynote presentations tomorrow morning at 9am ET along with my colleague @Jie Lou on our use of Prefect & Dask for machine learning in healthcare. Prefect (and Dask) have become very key parts of our technology and have delivered great results for our business. Paid tickets are required to attend live tomorrow, but I think the talks will get posted to the Dask Youtube channel for free access after the conference: https://summit.dask.org/schedule/presentation/62/dask-prefect-for-healthcare-machine-learning-on-aws/

is there anyway to pass a secret into an environment variable in prefect cloud? e.g. I know i can set env variables using a template or via the KubernetesRun object, but both approaches get displayed in plain text in the UI