hey all, I’ve also got a couple questions related to some issues we’ve encountered during some recent flow runs, not sure if this is the right place to ask for help: 1. we have a flow which at one point has 3 tasks that are triggered in parallel when their dependent task finishes. for whatever reason, two of those tasks got triggered automatically as soon as the parent task finished, but the other one was stuck in a pending state. it eventually started, but only after the other 2 tasks finished. is there any reason why the 3 tasks did not run in parallel? 2. it can sometimes happen that we attempt to restart a flow but we have to click the restart button a couple of times before the actual flow restarts. is there any reason why this happens or is it just a prefect bug? this has happened on a couple of occasions now (here i clicked the restart button, waited 3 minutes and then clicked again, and didn’t change anything between attempts)

Maybe I'm overthinking this, but I need to accomplish the following:

#input : Have a list of X: [X1,X2, X3 ...etc]
#input: total_times = 1_000_000
#property: x.do_work(1_000_000) = 1000 * x.do_work(1000)
#output needed: [X1.do_work(1_000_000), X2.do_work(1_000_000) ..etc]

@task
def long_running_task(x, n_times):
    result = x.do_work(n_times)
    return result

I have hundreds of Dask workers and want to split this work into units of

x.do_work(1000)

to get the output as fast as possible through maximizing parallelism. How should I write my mapping functions to achieve this? Do I just generate a list of

X_list = [x] *1000

for each

X

and a list of

iteration_list = [1000] * 1000

, then do

long_running_task.map(x=flatten(X_list), n_times = flatten(iteration_list))

?

Hi has anyone ran into issues with old versions of flows retaining their scheduled runs and not clearing it? The old version is archived and the schedule is off, but it will still have 2-4 jobs remaining that don't get canceled. This is causing us to run two jobs at the same time every now and then.

I'm trying to figure out how to set the output of a particular task in one flow as a parameter in another flow and I am just not getting it. I cannot find documentation or any code examples. This is my setup: 1. I have three dependent flows: flow1 -> flow2 -> flow3 2. flow1 has around 10 tasks. One of those tasks, say the 4th one, named GenerateRandomData, randomly generates a string based upon the data sent to it. So, something like this:

random_string = GenerateRandomData(data=data)

3. random_string from flow1 will then be a parameter sent to flow2, so the parameter is dynamically set at runtime since the value of random_string is different with each execution of the three flows. I learn best from reading code, so a basic example would be most helpful.

Hi, I'm running into an issue with the dask distributed executor where the scheduler is quickly running out of memory. The entire workflow is :

import prefect


@prefect.task
def do_nothing(n):
    pass


with prefect.Flow("Dummy Flow") as flow:
    listy = list(range(200000))
    do_nothing.map(listy)

The scheduler pod runs out of memory after around 300 tasks, the screenshot of the dask dashboard is attached. Has anyone ran into this issue or have any ideas for a fix? We normally run prefect 0.14.16 but I've tried on both 0.14.16 and latest (0.15.5) with the same results.

Hi, I have a few questions about managing different versions of flows: • Is it possible to get the flow's numerical version number and project name during a flow run, either from the

prefect.context

object or elsewhere? • The flow ID used in the URL https://cloud.prefect.io/paravision/flow/<ID> seems to be different from the flow ID that's output when registering a flow. Is it possible to get that ID during a flow run through

prefect.context

?

prefect.context.flow_id

looks like the same one that's returned when registering the flow, but would like to also get this other flow ID. • When registering a flow, it increments the version number and archives the previous flow version. Is it possible to un-archive a previous version so it can be run again?

Hello, I get a

FileNotFoundError, No such file or directory

while using Git storage to a private git server. I set Secret as ENV and specify

git_token_secret_name

. Does this mean that the agent can connect to the repository but don't find the Flow python file?

Does

StartFlowRun

support passing parameters as lists of dictionaries in Python? I am currently getting the following error when I try to do that:

prefect.exceptions.ClientError: 400 Client Error: Bad Request for url: <https://api.prefect.io/>
This is likely caused by a poorly formatted GraphQL query or mutation but the response could not be parsed for more details

I tried performing the same operation via the interactive graphql api and I needed to escape the

{

and

"

to get the query to work. Looking at the source it seems as though this doesn’t happen and might be the cause of my problem.

Hey peeps, What am I doing wrong here. I am trying to provide multiple clocks into my flow schedule ( trying with one to start off with ) but running into an error on the param. I have followed other examples, but it just isnt working here

schedule = Schedule(
    clocks=[
        CronClock(
            cron="*/7 * * * *",
            parameter_defaults=dict(account_identifier="inrun"),
            start_date=datetime.now(tz=timezone("Australia/Brisbane")),
        )
)

with Flow(
    name="betfair_flow",
    storage=Storage().in_a_s3_bucket(),
    run_config=RunConfig().fargate_on_ecs(cpu=512, memory=2048),
    schedule=schedule,
    executor=LocalDaskExecutor(scheduler="threads"),
) as flow:
    betfair_account_identifier = Parameter("account_identifier")

Error in 🧵

If I want to map over a task that takes several arguments. e.g.

def some_task(id:str, old: Dict, new: Dict)

and I have data as

[{"id":"some_id", "old":{...}, "new":{...}}, ...]

is there some easy way to explode the dict into the tasks, or is it best to write some wrapper task whose only responsibility is to take the data and pass it to the intended task.

Hello, If i have my python code on Machine 1, then make a flow and register it... then I launch an agent from Machine 2, would it be able to launch the initial flow? If no, what's the appropriate way to handle similar scenarios?

minor issue: just ran a flow which completed successfully, but the prefect slack app complained that the flow failed. any reason for this to happen/i should be concerned?

Run 

attractive-bug

 (

821098cf-c921-49ef-93d2-dd741b440aca

) of flow 

example

 failed 

STARTED_NOT_FINISHED

 SLA (

ee53f0cf-6d07-4dee-afb6-d1205baf141c

) after 60 seconds. See the UI for more details.

Hello, I am starting using prefect, I am a luigi user so I am testing the “Result/Checkpoint/target” feature in prefect.. I tried the simple ex.

import prefect
from prefect import task, Flow, Parameter
from prefect.engine.results import LocalResult

result = LocalResult(dir="./prefect-results", )

@task
def hello_task(named_params):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(named_params)
    return 'asd'

with Flow("hello-flow", result=result) as flow:
    named_params = Parameter("named_params",)
    hello_task(named_params)

flow.run(named_params='param')

I was expecting to find a file in the the

./prefect-results

folder.. the folder is there but it is empty.. I also tried

import prefect
from prefect import task, Flow, Parameter
from prefect.engine.results import LocalResult

@task(checkpoint=True, result=LocalResult(dir="./prefect-results"))
def hello_task(named_params):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(named_params)
    return 'asd'

with Flow("hello-flow", ) as flow:
    named_params = Parameter("named_params",)
    hello_task(named_params)

flow.run(named_params='param')

also empty.. Am I doing something wrong, or I misunderstood how this feature work?

Folks i am porting some of my Airflow jobs to prefect. And wanted to know, is FunctionTask behaviour is same as PythonOperator (Airflow operator) behaviour?

Hey all! I am looking for a way to use file targets for output caching, but parameterised by the upstream dependencies to a specific task. In other words, I don’t care exactly what the GCSResult file path looks like, but I need that 1) the filepath corresponds to that task’s upstream dependencies (such that one task could have many possible output results), and 2) all these different results could be cached at any one time. Has anyone done something like this? NB: what would be amazing is something like an

{upstream}

context option for a given task that somehow serializes the upstream dependencies into the result / target file name at run time. I know there is a

{parameters}

context value (so I could use this in worst case), but not every task in a particular flow depends on all of that flow’s parameters, so this would result in unecessary computation. Right now I implement something like this inside the task’s run logic itself, but I imagine it’s a helluva lot slower than letting the orchestrator deal with skipping tasks rather than entering every task to perform a file existence check before exiting (and also a lot less clean). Cheers for any help!

Hi all – I think I'm having a slow morning, but does anyone have a good heuristic around where to start debugging when you're trying to pass the result of one task to another, but your flow schematic is showing them as being unconnected? For example, I have something like:

@task
def custom_task_1(input) -> str:
  return f"Custom Input: {input}"

with Flow(...) as flow: 
    custom_task_1_results = custom_task_1("Hello World!")
    custom_task_2 = CustomTask2(message=custom_task_1_results)

but when I look at my flow's schematic, it doesn't show

custom_task_1

as being upstream of

CustomTask2

...I don't remember ever having run into this, so I feel like there's something quite obvious that I must be missing. I appreciate any help!

Hi all, I'm just getting started with Prefect and Docker as well. Basically, I'm trying to find out how I go from executing flow runs using the Local agent to using the Docker agent. I have a registered flow that's stored on Github. This runs fine using a Local agent in my local environment. Attempting to run the flow with a Docker agent fails with

ModuleNotFoundError

. To make this work will I need to create a docker image that includes all the dependencies? Also, this means that I will have to use another option for storage?

Is it possible to change the logging-level of parameter tasks? For flows that have many parameters (and most with default values), it'd be nice to remove them from the info-level logs, where they can be overwhelming.

Hey there! I am trying to set up checkpointing / result saving / output caching, and it seems everything is now working correctly when I run flows locally (I set

*prefect*.config.flows.checkpointing = True

somewhere in the code). But I can’t for the life of me get it to work when I use my Docker agent that’s connected to Prefect cloud. I set

--env PREFECT__FLOWS__CHECKPOINTING=true

as an env variable when I run the agent, but it never uses any checkpointing logic. Any ideas here? I’m more than happy to provide more info as needed

I'm seeing an odd bug when I try to reset a project (delete it and then recreate it). Here's a example of what I'm trying:

client.create_project(project_name="Hello, World!")
is_successful = client.delete_project(project_name="Hello, World!")
client.create_project(project_name="Hello, World!")  # FAILS HERE

The second call to create_project fails with a ClientError due to a uniqueness violation -- but in raising that error, the client is unable to tell which existing project causes the uniqueness violation (bc I just deleted it). Would anyone have insight as to what's going on here? I've tried sleeping between each line too so I don't think this is a race condition, but I'm not totally sure.

is there a trigger I can use to do some logging after a run is complete? I would love to write to a database when a flow finishes without having to add a task and muddle the logic of the DAG its self

Hey ! I’m running the prefect agent on AWS EKS (Kubernetes) and when firing a job (which creates a new pod on the same cluster) I get the error

NoCredentialsError('Unable to locate credentials')

from the job execution. In the past I used the ECS agent and I was able to pass a

task_role_arn

to it. However I see that both the

KubernetesAgent

and the run config for Kubernetes

KubernetesRun

don’t have a “role” parameter to inform. My next thought was that either the role attached to the

EKS cluster

or the

Fargate profile

should do the work, but as a test I gave admin to both and I still get the credentials issue. What am I missing ?

I have a task that globs for files given a file path, in my flow I call that task with something like

f"{os.path.dirname(os.path.realpath(__file__))}/layouts"

, however this causes the following (when executing from Prefect cloud):

Failed to load and execute Flow's environment: NameError("name '__file__' is not defined")

how come that global isn't available?

Hi! I am trying to run a flow with mapped tasks with a DaskExecutor on my machine, I have a couple of queries : 1. will the overall run time benefit if I run the same flow on Prefect Cloud? 2. Does prefect cloud manage the spawn of a Dask cluster resource ?

Privacy Policy Question: I'm familiarizing myself with Prefect to see if we want to use it. One thing I'm curious about is that while disabling telemetry in my config.toml, it looks like my browser is still trying to block screen recording on localhost:8080, used by logrocket. Looking at https://www.prefect.io/legal/privacy-policy/,

The LogRocket software has a feature that allows for the “video-like” replay of a previous user session

it looks like disabling telemetry might not disable logrocket's session recording? Am I interpreting that correctly or am I missing a config to disable it? If anything, logrocket seems more invasive than GA telemetry. Please let me know if this belongs in a different channel. Thanks

Hi all, I’m having some trouble with retrieving data from a child flow. I have a child flow register and that sits in folder A, the child flow is an ETL process that downloads and formats data from a S3 bucket. I want to retrieve the transformed data from the child flow after it ran. I am using create_flow_run to start the child flow and it seems to run, however when I use get_task_run_result to get the data from the child flow it just gets stuck. I’m not sure if I’m not passing the correct task-slug or if the problem is related to the context of the child flow changing. In principle I want to define a ETL process that I can use in other projects and just use the generated data from the child process as the data source in the new project I’m working on. Thank you in advance for the help!

Is it possible to know at runtime whether a Parameter has taken its default value or been overridden?

Hi everyone. I am trying to wrap the

Flow

class so that all tasks will report state to an external service. From what I undersand, subclassing the

Flow

class may not be a good idea as the prefect server will not use the subclass when managing the flows. I like that prefect has support for state change hooks (

state_handlers

), and I was wondering if I can use such a hook to modify the tasks of all tasks of a flow in order to add tasks state handlers that will report the task state change. Something like this:

def task_state_handler(task, old_state, new_state):
  report_state(task, old_state, new_state)

def flow_state_handler(flow, old_state, new_state):
  was_pending = old_state.is_queued() or old_state.is_scheduled() or old_state.is_pending() or old_state.is_submitted()
  if was_pending and new_state.is_running():
    for task in flow.tasks:
      task.state_handlers.append(task_state_handler)

Posted in #prefect-server

Hi ! , Is there anyone successfully submitted a flow in Openshift ? I was getting the prefect agent up and running in the cluster and try to execute the flow in it, but the state got stuck into 'Submitted for execution'. The debug log in agent didn't help much.

DEBUG:agent:Querying for ready flow runs...
[xx] DEBUG - agent | Creating namespaced job prefect-job-c4a440b5
DEBUG:agent:Creating namespaced job prefect-job-c4a440b5
[xx] DEBUG - agent | Job prefect-job-c4a440b5 created
DEBUG:agent:Job prefect-job-c4a440b5 created
[xx] INFO - agent | Completed deployment of flow run 549ced42-7840-43d6-89da-6f1a1f9cb378
INFO:agent:Completed deployment of flow run 549ced42-7840-43d6-89da-6f1a1f9cb378

Hi, Today, after using our CI/CD pipeline to push some changes to our PrefectCloud account, the information of the flow runs from earlier versions have vanished. The version numbers are right, but is like the olders never existed. Is this the intended behaviour?