was added to this conversation

Hey @Zach Angell 👋

Hey @Jeremiah 👋

👋

Welcome everyone! Sorry it’s a bit quiet while we get some things ready :)

Hello 🙂

Hello @Jay Leach !

Is this thing on?

Jamie we were hoping you’d lead a conversation on the details of the execution model

Hello 👋 Am I in the right place? @Jeremiah’s Twitter feed led me to believe this is a place for high quality pun content.

hey peeps, any preferences for organizing projects/namespaces based on prefect flows?

Introducing, the official prefect party parrot ---> marvin

Hi folks, I am starting to consider using Prefect in a project, but one thing remains to be clarified : is that currently possible to specify the worker ressource requirement for a task running on a Dask cluster? E.g.: I have a flow with two tasks A and B. I would like to specify that A should run only on GPU workers while B should run on NO GPU workers. I did not find anything about that in the docs.

Im trying to use Prefect with a dask kubernetes cluster (installed with helm stable/dask) - however whenever i install prefect on the dask worker pods it upgrades distributed and then the cluster falls apart - any one got prefect working in this way?

Hi, i am using imperative API's. in need of a clarification on using kwargs in overridden Task class run method. Am using two classes and i want to pass kwargs to one and the modified kwargs to the next task. Am not sure if i had done right in putting pieces together but am getting "TypeError: too many positional arguments" from the second class's run method. I have attached the code as well. Thanks in Advance

heyyo general design question about boundaries between tasks and level of separation of concerns and idempotency... I am writing my first project with Prefect and have come up on a situation where I want to pre-process a bunch of data in one task before persisting it in another task

but in the task 2, I need to do post-processing on the selfsame data structure as outputted by task 1.. in airflow it would seem this is a no-no without persisting it somewhere first.. I guess just wondering: is there is a general preference for mixing purposes and keeping tasks orthogonal?

Hey all! A question I couldn't find addressed in the docs: is there a way to run a flow from a particular "entry point"? E.g. if you have a flow with sequential tasks A -> B -> C -> D, could you invoke the flow from task B or C? A concrete use case is a data processing pipeline where task A is time-consuming and most of the time doesn't need to be run.

@Chris White Regarding state and metadata (e.g. log of task runs), you mentioned a while back that Prefect Cloud will have this setup by default. If we wanted to roll our own DB, how custom is this kind of setup and is there any Airflow-like equivalent to setting up your "metadata db" with a script?

Hello, I am running through the Docker Prefect example (https://docs.prefect.io/guide/examples/imperative_docker.html) and it says that it is using the Imperative API to build out the example. Is there a reason why the Functional API couldn't be used for this purpose?

Hi there, I was trying to get the dask scheduler UI to be a bit more expressive. Normally, I have all my running tasks grouped under the run_task and run_fn names, but what I would like is to have the real task name. I managed to extend

prefect.engine.executors.DaskExecutor

to add the task name as the

key

parameter sent to the dask client:

import uuid
from typing import Any, Callable

from distributed import Future
from prefect.engine.executors import DaskExecutor


class MyDaskExecutor(DaskExecutor):

    def submit(self, fn: Callable, *args: Any, **kwargs: Any) -> Future:
        if 'key' not in kwargs and 'task' in kwargs:
            kwargs['key'] = kwargs['task'].__class__.__name__ + '-' + str(uuid.uuid4())
        return super().submit(fn, *args, **kwargs)

So far this is working as expected, with the exception of tasks that use the

.map(...)

idiom, where the run_fn still appears with that name. I have looked further into prefect’s code but it seems that to achieve this I would need to create my own task runner. Since that seems a bit complicated, I just wanted to ask if there is another solution or if this is just a bad idea?

this is a super interesting idea! I love it

for mapping, you should look at the

map

method of the

DaskExecutor

-> when we map tasks, we use

executor.map

instead of

executor.submit

I tried that, but the thing is that

DaskExecutor.map

does not receive

kwargs

, and I am using the “task” kwarg to guess the task name.

ah i see

ah

inside of

DaskExecutor.map

, try:

task_name = prefect.context.get("task_full_name", "some-default-value")

👍 That works for me ! Thanks! My only regret is that my custom executor now repeats the

DaskExecutor

map code, because I cannot do

super().map(...)

given that the executor does not admit kwargs. I can live with that, but I will have to keep an eye on any new changes on that class on future versions.