What would be the cleanest way to write a brief description about what a flow does in the UI?

Hi, trying to get only get the error logs from graphql. Does anybody know how I can restrict the query below so I don’t get all the logs levels?

query {
	flow_run{
    logs {
      message
      level
      info
      created
    }
  }
}

Of course a totally different query to get error logs from graphql would also be welcomed

What’s the difference/relationship between a flow group id and a version group id? I’ve noticed that as a republish a flow I see the flow ID change while the flow group id and version group id stay the same

Hi, how do i toggle local/cloud get of PrefectSecrets? I set a few keys in the cloud UI but I am unable to retrieve them with PrefectSecret currently

Is there a process to suggest a field get exposed in the graphql api? If so, I wasn’t able to see duration for flow runs as an available field, I understand it can be derived from start_date and end_date, but just wanted to put this out there

Hi - looking to pass AWS credentials to a Kubernetes agent at runtime so as to interface with the pynamoDB python library. I’m getting the following error when I attempt to scan a table in DynamoDB:

Failed to scan table: Unable to locate credentials

I have

AWS_CREDENTIALS

stored in Prefect cloud as a Secret, and I’ve tried passing credentials as envs passed to KuberenetesRun:

env={
        "ACCESS_KEY": Secret("AWS_CREDENTIALS").get().get("ACCESS_KEY"),
        "SECRET_ACCESS_KEY": Secret("AWS_CREDENTIALS").get().get("SECRET_ACCESS_KEY")
    }

Returns the same error. What’s the best / preferred approach to authenticate a Kubernetes agent against AWS?

What options are available for storing and running flows that span multiple files (e.g., flow.py, flow_tasks.py)?

Hello everyone, I am currently looking to try out a self hosted Prefect Server and UI instance on AWS. Based on the services I saw in the docker-compose file over here: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/cli/docker-compose.yml Has anyone attempted to deploy this on an ECS Fargate cluster ? Apart from the Postgres instance, I would like to deploy each service into its own Fargate instance. I am just wondering what could be some sane cpu and memory specs to kicks things off assuming I have a few flows with with an overall 1000ish tasks running twice every hour. I would also like to know if all these services are stateless, meaning I can scale them horizontally via a load balance ? Or are there any services here which can only be scaled vertically in case of performance bottlenecks ? Any help would be appreciated ! 🙏

Hey all, Im running a flow using VertexRun - It has multiple steps and it runs perfectly. Once a week at a specific time I’d like this flow to invoke another flow that will train a model. I used this code -

if pendulum.today('America/New_York').weekday() == 2:  # Monday is 0 so Wednesday is 2
        x_flow = create_flow_run(flow_name=PREFECT_TRAIN_FLOW_NAME, project_name=PREFECT_TRAIN_PROJECT_NAME)
        wait_for_flow_a = wait_for_flow_run(x_flow, raise_final_state=True)

This code is of course inside my “with Flow….” code block. Now when running this code alone (in a dummy flow) - it works and x_flow gets invoked. But when running this code in my real flow, after some other tasks - nothing happens. I cannot even see the task of “wait_for_flow” in prefect cloud(flow -> tasks tab) - seems like its getting ignored. What am I missing here? Thanks

Hello! Is there a way to run flows solely on EC2 agents without using ECS? I've seen some past messages about running flows on EC2 with ECS, but my team is looking to only use EC2 and spin up/down instances based on a custom image on our end.

Just want to say I have been focusing on getting prefect examples ready to show my team and after a few lessons learned it all worked like a charm. I am excited to be on the cusp with using Prefect to manage our data pipelines. The reporting and graphs already are way more than I expected. Thanks.

Hello, I have a situation where I am generating tasks at run time based on some configuration (so the number of tasks may be different depending on config). I would also like these tasks to be able to be run in parallel using the local dask executor. The difficulty I'm running into is that I need the tasks to be defined individually in the flow in order for the local dask executor to be able to run them in parallel. I do not know how many tasks there will be until runtime, so I cannot define the flow with individual tasks. I cannot simply loop through the task functions in the flow because the executor will not execute the tasks in parallel. Is there a recommended way to accomplish this?

Dear Prefect-Community, My Senior send me here to politely ask if it is possible to map prefect-task to individual kubernetes pods instead of paralell python-tasks. :)

I've been playing with Prefect Orion and I'm wondering how I can view the logs associated with a run in the UI?

Hi! How i can import my custom library to the flows run through kubernetes agent?

is there an equivalent to task.map but runs the mapped parameter elements in serial instead? basically identical to map but forced to run in order. I looked at the loop signal as an option, but I don't think that will lay out the schematic in the UI as nicely as map does

Hello all ! I'm looking for a simple way to modify a PrefectResult, in order to re-run a task with the correct input without re-running the whole flow. Is there a way to do it in Python, or otherwise with a mutation in GraphQL? I explored the schema but could not pinpoint where to find the result given its "location" id 🤔

Hi all! This may be more of a Docker SDK question, but was hoping a friendly prefecter might have encountered this issue. I'm trying to bind a Windows network path for my flow's run_config. I read the Docker SDK docs on create_host_config, but not sure what I am supposed to change. For example:   1) I can use the code below to successfully bind a local file path (for read/write) to a Docker container   2) However, if I use a UNC path to a network drive, I receive a "Server Error" that the path is not a valid Windows path Any help is welcome, thank you!

### THIS CODE ALLOWS ME TO BIND A LOCAL FILE PATH
from prefect.run_configs import DockerRun
import docker

  flow.run_config = DockerRun(labels=['my-label']
                            ,host_config={'mounts':[docker.types.Mount(target='/public'
                                                                        ,source=r'//c/some/local/path'
                                                                        ,type='bind')
                                                ]
                                        }
                        )

### THIS CODE THROWS ERROR 
  flow.run_config = DockerRun(labels=['my-label']
                            ,host_config={'mounts':[docker.types.Mount(target='/public'
                                                    ,source=r'\\path\to\windows\network\shared\drive'
                                                                        ,type='bind')
                                                ]
                                        }
                        )

👋 hey all, if there's anyone new to the community hanging out here, @Kalise Richmond and I are going to go live on Twitch in 5 minutes and go from 0 to first flow. See you at twitch.tv/prefectlive!

not wanting to disrupt the stream, do you guys have any timeline to prefect orion? looked at that and I'm wanting to start moving legacy code into prefect but I love how orion looks and didn't want to start moving and then having to deep dive again to change it once more ^^

How do I create and run flows on a schedule in Prefect Orion? Let’s say I have this:

schedule = IntervalSchedule(interval=datetime.timedelta(seconds=10))
deployment_spec = DeploymentSpec(name="hola", flow=greetings_flow, schedule=schedule)

Do I now use OrionClient and some methods to deploy? We currently use Prefect core in a pod and simply do flow.run() with a schedule attached, and I’m looking for a working example to do the equivalent in Orion (even if I gather behind the scenes it will do something else via a server and DB). Basically I’m looking for this: https://orion-docs.prefect.io/concepts/deployments/#running-deployments-with-the-api 🙂

Hi everyone, are results returned from the API sorted by

created

(or any other field) guaranteed to be returned in the same order? Thanks!

What's the easiest way to specify to run N simultaneous extracts with Dask? E.x. each extract is like this:

df_2015 = extract_past(connection, start_date="2015-01-01", end_date = "2015-12-31", task_args={"name": "Extract 2015"})

I define the Flow that is started with multiple starttime and at the specified time, i wanna to run the specified task. how to do that? tks

I have a problems with task that handle reading files from ftp server. After that, I insert data into database. That's exception: Plz give me solution

Hi, i want to use docker compose for prefect like we are using for other services. Is there a way so i can store everything like custom modules and flows in docker image and when i run the docker compose i should be able to access and run the flow using prefect cloud

Is there an elegant way to control the level of concurrency using the

map()

functionality? My scenario is a list of URLs that should be used for making requests to but want to perform the requests in batches of 'n' rather than overwhelming a server with them all at once.

we’ve got a large DAG with the same task repeated in several places using different inputs (do X for A, do X for B, etc), and it can be hard to quickly understand which task failed or even just understand the structure of the DAG itself. I tried using both the

task_run_name

and

name

args in the

task

decorator to provide more context, but neither works the way I’d like it to. as a MWE, here’s a sketch of what I’d like to happen:

with Flow() as flow:
    task_a(entity="foo")  # shows up as "foo_task" in the DAG
    task_a(entity="bar")  # shows up as "bar_task" in the DAG

the issue is that

name

only takes in static strings (so

name="{entity}"

doesn’t work) and

task_run_name

only sets the task run name, meaning it will never show up in the schematic outside of flow runs (even in flow runs, the name is only shown when clicking on each individual task card, making it hard to see an overall picture). is there any way to achieve what i’d like to do?

Can we write a task that could return the list of files/folder from the result storage that prefect is using. The idea is that the flow is setup with S3 storage this would do an s3 list_object and a local would a recursive scan. This task could take a prefix to Supports filtering A speudo code could look like that

@task
def list_file_from_storate(prefix: str) -> List[str]:
    result = prefect.context["result"]
    return result.list_files(prefix)

Looking at the context doc I cannot see the result been available here https://docs.prefect.io/api/latest/utilities/context.html We we want is to be able to interact with the result using task from the flow setup.

Hello guys, I have a weird behaviour while using prefect Orion deployments with the UI, I have this deployment file:

import pandas as pd
from prefect.deployments import DeploymentSpec
from prefect.orion.schemas.schedules import IntervalSchedule

from pred.predict import predict_flow

DeploymentSpec(
    flow=predict_flow,
    name="Prediction"
)

I create the deployment without a problem and I can see it in the UI. I also can run it successfully from CLI using:

prefect deployment execute 'Prediction flow/Prediction'

But When I try to run it using the “Quick Run” in the UI, a “Scheduled” task is created and is never run. I can see it in the lateness graph. one task didn’t run for over an hour during my lunch break. Same happens if I add a schedule to the deployment. any ideas why that happens or how to fix it?