Hi, I am running a flow using DaskExecutor, after the Dask cluster starts up successfully, the flow encounters an error and goes FAILED, when I rerun the flow I get the following error.

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"services \"foobar\" already exists","reason":"AlreadyExists","details":{"name":"foobar","kind":"services"},"code":409}

It seems that the error occurs when I try to create a service with the same name. How can I rerun the flow when using DaskExecutor?

Hello folks, is it fine/possible to use slack notifier from API functions (

prefect.utilities.notifications.notifications.slack_notifier()

) as a state handler for Flow?

We need

ResourceManager

`map`ped to list of days. We have a task, archiving data for one day. The task is using temporary dir (mapped to shared memory) and we use ResourceManager to cleanup the tmp dir when a day is processed (there is day subdir per processed day). Now we want to run the same for list of days and we struggle with mapping days to resource manager.

hey community 👋 🙂 , Prefect is one of the option to adopt inside the company, but we would like to resolve our use case. A simple batch process like running a job in a EMR cluster, which would be the approach using P? Airflow has some good adapters, but I am curious if we could do the same using prefect

Hi, does anyone know how to set the service account for the dask workers in an ephemeral cluster? This may be more of a dask question but the k8s pods are being set to

default

and was wondering if anyone ran into the same issue. Example executor code is in the thread.

I'm trying to configure Run schedules for flows using the API. I can't find any documentation for it. Nothing in the CLI or anything. Any pointers on where to look?

Hi! Does anybody know what’s the best way to initiate/run prefect job from gitlab ci/cd pipeline?

I am writing a small Lambda function to trigger a prefect flow run on demand. I was wondering if anyone has done similar and could let me know if these queries look right • given a flow name, get the latest version’s id

LATEST_FLOW_BY_NAME = gql(
    '''
        query LatestFlowByName($name: String) {
          flow(
            where: {name: {_eq: $name}},
            order_by: {version: desc},
            limit: 1,
          )
          {
            id
          }
        }
    ''',
)

with:

variable_values={'name': name}

• trigger a flow run

CREATE_FLOW_RUN = gql(
    '''
        mutation CreateFlowRun($input: create_flow_run_input!) {
            create_flow_run(input: $input) {id}
        }
    ''',
)

with:

variable_values={
    'input': {
        'flow_id': flow_id,
        'parameters': parameters,
    },
},

My goal here is to specify as little as possible, to ensure the flow runs with the defaults configured for it. I.E. I don’t want to muck with the RunConfig, etc. I am mostly concerned about my logic for getting the latest flow run. I tried to grok the flow groups / versions tables but did not have much luck.

Hi, I was wondering if it is possible to grab/filter by a flow parameter’s value via a GraphQL query. I’ve been messing around with the interactive API and can see you can filter

flow_runs

by parameter by using something like:

flow_runs(where:{parameters: {_has_key: "process_date"}}

but what I would like to do is filter the query for any flows that ran on a specific process date, such as, 8/17/2021, rather than filtering for flows that have the

process_date

parameter. Thanks!

i’m not sure if this is a support question or a feature request: i’ve got a flow of flows set up, meaning that a top level flow kicks off a DAG of other flows i have created. my issue is that i can’t easily access the flow run of the subflows from within the page for the top-level flow run. i can click on

person-build

in the schematic view, but that just takes me to the local (“zoomed-in”) view of the task itself, i can’t see the DAG for the person-build flow run. the only workaround I’ve found is to inspect the logs and copy the link to the flow run from there. am i missing something obvious or is there no easy way to access a subflow’s details? this is basically our setup:

person_build_flow = StartFlowRun(flow_name="person-build", project_name=get_stage(), wait=True)
release_flow = StartFlowRun(flow_name="release", project_name=get_stage(), wait=True)

with Flow(
    "build-then-release",
    executor=LocalDaskExecutor(num_workers=8),
    result=PrefectResult(),
) as flow:
    release_flow(upstream_tasks=[person_build_flow])

Hi there. I just had something strange happen on Prefect Cloud and was wondering why. I have a flow that was running every night. Another person registered a new version of the flow (version 9). Until now, I had been the only person registering this flow. Before the new flow version was registered, I could see the run history but once the new version got registered, I cannot see the history anymore. Also, I don't see the prior versions in the version drop down.

Hey all, this is more of a front end observation. So on an actual day-to-day basis, if the front end is used for monitoring, I'm worried it will get exceedingly hard to tell at a glance what is scheduled and what has run, as it all seems to be color coordinated in the same way and the jobs are all obfuscated behind run pseudonyms. I'm currently running different schedules for just one pipeline, and am already not sure which one was which without clicking down to the parameters in each individual flow. Guessing at which one was which based on the height of the time-taken-to-run bar isn't exactly helpful either. Any thoughts on how to make things more clear?

I have a state handler that checks for is_failed() and sets the state to Skipped if that is the case. This is a requirement. Something like this

def handler(task, old_state, new_state):
    if new_state.is_failed():
        // do stuff here
        return Skipped()
    return new_state

the consequence of this is that the flow finishes with a SUCCESS state, even though a task actually failed, but I need to show the flow as FAILED. To solve this, I created a new flow that calls a different state handler and checks its state. If the state is skipped, I return Failed() from the handler. Something like this:

@task(name="final check", state_handlers=[final_check_handler])
def final_check():
    pass

and the state handler

def final_check_handler(task, old_state, new_state):
    if new_state.is_skipped():
        return Failed()

This accomplishes what I want, but surely there has to be a better way.

Is it possible to send PandaSerializer results to a flow triggered by StartFlowRun?

Hi I have a flow represented by the DAG below. I ran it locally with

flow_result = flow.run()

and visualized it with state. The log says

Succeed: all reference tasks succeeded

but I can see a failure right there in the stateful DAG. Also it doesn't show the full DAG with the downstream tasks!!! I have also attached the output of

flow_result.result

and it doesn't show the

get_duplicated_pairs_info

task. Any tips on what went wrong and how I can debug from here? (Full log in comment)

Hi folks! I am trying to use Prefect, and found that it is rather slow for my use case. I thought it might be a mistake I made, but I've created a simple reproducible example that illustrates the issue. Basically: it seems context creation at the start of

task_run

is what is causing a huge amount of overhead. I would expect context creation to be very light weight... more details in the thread below.

Hi All, Is there away to configure the dashboard to use colors other than red & green? We have a color blind individual who can't distinguish red from green. Thanks

Hey! I'm currently following this guide by George Coyne from Slate to set up dbt orchestration with prefect. Does anyone have suggestions for other guides I can read into? My biggest roadblock right now is being able to pull the dbt repo on GitHub, would love any tips y'all have!!

Folks, i am triggering a flow run via local prefect server (through UI - “quick run button”) but it keeps scheduling (shown yellow dot in graph) and never runs. Task state always shows “pending” any idea what could be wrong?

i have registered the project as well.

screenshot:

the flow runs if it do

flow.run()

in code but triggering from UI is not working 🧐

I tried universal IT solution “*Did you turn it off and on again*” - did not work :-)

on cloud the behaviour is same. this is my code snippet:

with Flow(
    name=flow_name,
    # schedule=schedule,
    state_handlers=[slack_notifier(only_states=[Success]), flow_state_handler],
) as flow:
    ---some logic---
        for sync_type, cmd_parameter in sync_commands:
            ---some logic --
            sync_command = (
                f"{command_prefix} {source_bucket} {destination_bucket} {options}"
            )
            s3_sync_task = ShellTask(
                command=sync_command,
                return_all=True,
                name=task_id,
                slug=task_id,
                timeout=timedelta(minutes=task_execution_timeout),
            )()

            # Create a task to validate sync commands.
            validate_s3_sync_task = FunctionTask(
                fn=validate_s3_sync,
                name="".join(["validate_", task_id]),
                slug="".join(["validate_", task_id]),
            )(
                timeout=timedelta(minutes=task_execution_timeout),
                s3_sync_command=sync_command,
            )

            # Define dependencies between tasks.
            s3_sync_task.set_downstream(validate_s3_sync_task)
            
# flow.run()
flow.register(project_name="demo-project", labels=tags)

hi, i’m trying to restart a failed flow of flows, but restarting it always results in a not-very-helpful error. [log in thread] i’ve tried restarting the flow several times (and usually have to hit the restart button twice before it actually submits the flow for execution) and i always get the same error. the logs for first dependent flow (person-build) don’t pick up anything regarding the restart for the parent flow. this is essentially the setup i have for the parent flow: [code in thread]

hi, i have a huge list of latitude and longitudes (over 1MM) and I am trying to download images corresponding to each of these lat-lons and do a bunch of downstream processing on them. I am using a

map

to parallelise over the lat-lon pairs but i have a couple of questions here: 1. I know if i use a

DaskExecutor

or

LocalDaskExecutor

the flow is distributed, but is there any limit to applying

map

over such a large collection? 2. instead of using threads to run the computation, is it possible to make use of

async

as the tasks (most of them) are heavily IO bound? What are some considerations I should make here? Thanks!

Hiya, IIUC agents poll the server every 10 seconds. Are there any plans to decrease this latency and enable flow runs to start immediately once scheduled?

hey guys, I’m facing a kind of an obscure error using running a flow with GitHub storage…it seems like I’m downloading the files successfully, but then the flow fails for not finding a certain file…is there a way to produce more verbose logs so I can know exactly what file it fails on?

When looking at the Cloud dashboard we don’t see all the currently running job and it makes it kind hard to monitor what’s going on. I think this relates to long running jobs that are interspersed with shorter jobs. The start of the job gets pushed off the side of the timeline.