Hey, We use CircleCI to test and register our flows (Prefect v0.14.12) which we use Docker storage for. In our recent attempts to register a new flow we have been hit with this error and failing to build:

docker.errors.DockerException: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory'))

(full trace in this thread) From reading around is looks like a potential docker issue, an inability to connect to a socket?! Has anyone else been hit with this?

Hi, while trying to execute a task that connects to a db(mongo) and extracts a collection I get this error :

Unexpected error: TypeError("cannot pickle '_thread.lock' object")

The task blueprint looks like this :

def mongo_connect():    
    db = get_db() // wrapper on top of pymongoclient to connect to the specified db
    collection = get_collection(db). // wrapper on top of pymongoclient to get a specified collection form the db
    return collection

I understand that prefect needs to pickle all the tasks , but this code works fine while being run as an independent script (without prefect decorators). How can I make the connection threadlock safe or rather pickle safe?

Can a flow be registered without storage configured? Let’s say I want to use docker storage, but I have a separate build and push workflow that I’m using and then just pass the image to my run configuration

Hello everyone 🙌 I'm having a problem with scheduling flows, all of my flows did not schedule automatically overnight and I can't activate them through the interactive API (When I try to

set_schedule_active

I always get

"success": false

). When I try to manually toggle the flow through the GUI the entire server hangs until I manually kill the GraphQL process on the server. Can anyone help or provide some insight? I'm acting as an interpreter for the tech team so I have to do some back and forth with my questions.

Is there a way to specify an alternative location for

~/.prefect/config.toml

? Maybe an environment variable, because I don't see any CLI flags about it?

Hello! I was wondering what are some of the patterns that people use to set up flows with configurable number of workers for Dask? Given that you need to pass the executor to a flow it seems like you can't use parameters (as far as I can tell)? Right now we're using KubeCluster to spin up ephermal Dask clusters in K8s but our n_workers is hardcoded, but we'd like to configure it per environment or possibly based on task inputs in the future. Thanks!

Does anyone have any experience with dynamically configuring a state handler for a flow or some subset of a flow's tasks? We have a flow that basically serves as a "flow template" (specifically, for running different

dbt

projects) that multiple teams at my org use. The individual teams will set up an orchestrator flow that has one task (the

StartFlowRun

task), and they'll pass in the various parameters that they need to pass in so that their specific

dbt

project will run. They also will directly attach state handlers that they configure to their

StartFlowRun

task within their orchestrator flow. Of course, in order for the status of the underlying template flow to bubble up to the orchestrator flow, we have to set

wait=True

on that flow. The issue with this setup is that it unnecessarily eats up our flow concurrency, since this setup always leads to at least 2 flows running at a given time. We don't really need the orchestrator flow to wait around; we're only keeping it waiting as a means to capture the template flow's status, which then gets passed along to the custom state handler that the respective team within our org has configured.

Hey Team! it seems like the interactive GraphQL API doesn't reflect my role permissions. I created a tenant using GraphQL, then tried to assign it my license with the assign_tenant_license mutation. It's forbidden; then i just created my tenant through the UI and it worked, automatically attached with my enterprise license. Now i have to delete the old zombie tenant but i don't have permissions to do that either 😞 . Any resolution to this?

For a Cloud Automation to a webhook, it mentions that you can use a Prefect Secret for the Webhook you want to send. But I cannot figure out how to get the secret that I made to be filled in here. Anyone have any ideas?

Should

prefect register -p my_flow.py

be updating the version of the Flow? I have changed the code and ran

register

again, multiple changes and multiple registers, but the VERSION is still

1

in

prefect get flows

Hi Guys, We're looking to use Webhook Storage to store the flows in our local environment. The Prefect server will be hosted in a K8 Cluster, any recommendations for a storage service that would fit that requirement?

Hello! when I see the calendar on the board it appears: Reminder! The Prefect Scheduler will only schedule 10 runs in advance. How can I program more scheduling??

Hey, I have been trying to delete a project in the UI for half of the day now. Keep getting this error

Hi everyone, I am new to prefect, so maybe my question seems naive! is it possible to use @task for methods inside a class? I have a class with several methods that can be set as task.

Hello folks, i am trying to find a way to give a flow runs a name (currently its generated dynamically in prefect, see screenshot) Is there a way i can configure a flow run name in a code? I don’t see any parameter in

flow.register()

I looked into the documentation but couldn’t find it.

Hey there! In the Prefect settings, you can use Parameters with the data type 'list', 'dict'. Tell me, what is the maximum number of elements a list and a dictionary in a parameter can contain?

Hello there! If there a way to perform a mapping with a specific batch size ? for example lets say I have two tasks

f

and `g`:

ys = f.map(xs)
zs = g.map(ys)

Is there a generic way to give

g

batches of y values, for example 10, rather than one at a time ? (And obviously, without having to wait for all the ys to be computed.) Or is this something that is specific to the executor used ?

Hello Prefect. I have a question about dealing with Flow Parameters. Whats the preferred way to concatenate flow parameters that will be passed into a number of task? Let me type up a simplified example below

Did the "Mapped Tasks" panel go away/move recently?

Hi, does anyone know if the env vars set on the prefect agent (deployed in k8s) get injected into a dask cluster automatically? Or do we also have to duplicate them when making the pod spec for dask? Our executor is defined like:

flow.executor = DaskExecutor(
    cluster_class=lambda: KubeCluster(
        name=f"dask-{flow_image_name}",
        pod_template=make_pod_spec(
            image=prefect.context.image,
            cpu_request=worker_cpu,
            memory_request=worker_mem,
            threads_per_worker=threads_per_worker
        )
    ),
    adapt_kwargs={
        "minimum": min_workers,
        "maximum": max_workers
    },
)

We have a CI/CD pipeline that automatically registers flows for us (e.g. when a PR is opened); in my registration script I do something like:

if not args.dry_run:
        logger.log(SUCCESS, f"Registering flow {flow.name}")
        flow.register(project_name=args.project, idempotency_key=flow.serialized_hash())

Everything is working as expected, except I'm getting more versions of the flow (after multiple registrations) than I was previously expecting. It makes me wonder what I don't understand about the

idempotency_key

- could someone explain to me what could cause the serialized_hash to change unexpectedly? for example, if a different VM registers flows, would the hash change?

Hey there. I’ve been playing around with Docker storage today, trying to get all source code packaged together with the flows each time they are registered, and am using the

files

and

env_vars

attributes as outlined in the Docs. But it seems that my

.dockerignore

file (in the directory from which I am registering the flow) is ignored by this build process. I have massive data directories nested at different levels inside the codebase so it’s essential there be a way to include this somehow. Anyone come across this / have any ideas? I’m curious about playing around with a custom Dockerfile, but the code that copies flows to a tmp Dockerfile in

create_docker_file

is dynamic so it seems tough that I could write a standalone Dockerfile to handle this case.

Hi there! I have some (not all) tasks that need need PandasSerializer, and also need to use different result types depending on if dev or production. I have a solution for this (below), but wondered if anyone had a cleaner solution?

condition = True
resultType = LocalResult if condition else S3Result

@task()
def create_list():
    return [1,2,3]

@task(result=resultType(serializer=PandasSerializer(file_type='parquet')))
def to_df(a_list):
    return pd.DataFrame({'col1': a_list})

with Flow(
        name="my-flow",
        result=resultType(serializer=PickleSerializer())
) as my_flow:
    my_list = create_list()
    my_df = to_df(my_list)

Hey prefect team! I'm trying to register multiple flows generated in a loop in my file. However, it seems to only register the latest flow in the loop. Whats the right way to do this?

Is there a way to pass indexed items as arguments to a mapped tasks? E.g. If I have some inputs that are a list of tuples that i'm mapping over can I specify specific elements to pass as args when mapping over them? E.g. Something like this:

@task
def gen_records():
    return [('item', 1), ('item', 2), ('item', 3)]


@task
def subscript_input(x, y):
    return x+str(y)


with Flow(
    name="Test Flow",
) as test_flow:
    records = gen_records()
    total = subscript_input.map(x=records[0], y=records[1])

Hi, i have a flow with docker storage in ECR and a KubernetesRun with image_pull_policy Always , i have an agent running on my EKS cluster, for some reason when i change the flow code and builds a new image and run the flow from cloud, the flow runs with the old image, i validated this when i change the image tag then the new code is used, any clues how to solve this?

Hi, if I am trying to run a heavy transform task on a large data frame -

@task
def transform(df):
   return transformed_df

with Flow("transform-1") as flow:
   transform(df).                 # current implementation
   for each df in filtered_dfs:   # suggested way to utilize paralellization optimization on tasks
      transform(df)

1. Will breaking down the data frame into small chunks and the calling the transformation task on each batch of the dataframe help me reap the benefits of prefect parellelization using a dask executor? 2. Given that the input to a transform task is a large dataframe - what other steps can I consider to optimize the turn around time of the flow ? (suggest using different run configs and prefect paradigms of writing the code)

Hey prefecters... is there a mechanism for mapping task args to a map? Specifically trying to set the name of the task... you can see from the example below only the static name seems to work...

dirs = ["dir1", "dir2", "dir3"]
commands = [f"rm -rf {d}" for d in dirs]
names = [{"name": f"removing dir {d}"} for d in dirs]

shell = ShellTask()

with Flow as f:
    shell.map(command = commands, task_args={"name": "removing dir"})
    # shell.map(command = commands, task_args={"name": dirs})          # not working
    # shell.map(command = commands, task_args=names)                   # not working

f.run()

Possibly not getting some syntax right. Thanks.

Hi, I’ve opened https://github.com/PrefectHQ/prefect/discussions/4935 to discuss data lineage if anyone is interested

Follow up from my message above about the

DaskExecutor

not getting the env vars set in the run config, this has caused other issues in the way we have been deploying flows to different environments (test, dev, prod, etc). Previously, using the

LocalDaskExecutor

, we could set the executor on the flow itself and it was generic enough to work across all of our environments. Now we also need to change the dask config depending on our target environment, but it seems like setting the executor outside of the

with Flow()

definition just gets ignored. I believe this has something to do with the

storage

and

run_config

being serialized with the flow but the

executor

is not right now. An example flow and registration script is attached inside this thread. Does anyone know of a work around for this issue or have a better way to do this?