I’m trying to dynamically name a flow following these instructions. But is there some way to name the flow run with the labels used or the environment variables present in the flow run?https://github.com/PrefectHQ/prefect/discussions/3881

Hey guys, is this setup possible? I already a have

with Flow("dummy_flow") as flow:
   ... # parameters, logic, etc

I want to re-use this same flow (create another one), but reuse the logic of the previous one without have to copy-and-paste the old one is that possible? I want to create a second flow ex: "second_dummy_flow" but using the same logic of dummy_flow, so on prefect ui i will see 2 flows that will do the same thing

I’m working on a flow that will fan-out/fan-in in a dynamic manner. For each path, there are linear steps that are chained together. I’m taking advantage of

map

and have been successful with it previously for things like

create_flow_run

and `wait_for_flow_run`; however, the

wait_for_flow_run

input arguments exactly match the output from

create_flow_run

. I’m trying to think of how tasks can be chained together where the tasks take mapped arguments from upstream but it’s not included in the direct parent’s output. For instance, for several tasks I to prepare/process/finalize something, I need the same configuration information for multiple thats was a result of a

discover

task; however, this configuration information is not returned all the way through. I’m curious what would be a good approach to something like this. If I have an original fan-out that say returns a list of 5 results that get mapped and tasks keep returning different lists of 5 results is it safe to pass these results downstream and match them up by index order?

# pseudo code
five_config_results = discover()

five_state_results = prepare.map(five_config_results)

five_process_results = process.map(
    five_state_results, config_results=five_config_results
)

five_finalize_results = finalize.map(
    five_process_results,
    five_state_results=five_state_results,
    config_results=five_config_results,
)

Or would another approach be needed to capture the matching results from each task to give to others?

Is there any way to preform a rollback on a flow (to a previous version) using the GraphQL api?

Quick Question on Orion, after the recent upgrade to 2.0a10, I have seen that Orion flows work fine whenever I run but, I don't see anything on the dashboard 127.0.0.1:9090. Not sure if I am doing something wrong, anyone?

Ever since one of the recent updates (either 0.15.12 or 0.15.13, I'm not sure which), the machine running the server seems to accumulate zombie processes at a stunning rate. It ends up with hundreds in a few hours, thousands in a day or two. I thought I'd seen an issue related to this at some point in the Github issues list last week saying the issue was being worked, but I can't seem to find it anymore. Is this still a known issue being tracked somewhere? Or is there a fix for it?

We are using S3Result stores and configure our flow's storage in the constructor like this:

result=3Result(bucket=s3_results_bucket)

The flow's output looks like it is using a default template something like

YYYY/MM/DD/**.prefect_result

. I would like to reuse this bucket across flows and environments, with a template something like

{environment}/{flow_name}/YYYY/MM/DD/**.prefect_result

How can I accomplish this?

whats the best way to get the most recent failed flow runs for a specific flow in Python?

Hi all! I've developed a Flow that has multiple Clocks on its schedule. Each Clock has its own parameters for the Flow and labels. I was wondering what would be the best way of easily identifying runs for each of the Clocks on the UI. I found this closed PR on the github repo that would help a lot: https://github.com/PrefectHQ/prefect/pull/3664 A colleague has suggested refactoring the Flow as a Task and then declare multiple flows with different names, but this doesn't sound right... Do you have any recommendations on this?

How can I use *`prefect.utilities.gcp.get_storage_client` for

Flow(...*storage=GCS(bucket=...

? I'd create a client myself and without using

GOOGLE_APPLICATION_CREDENTIALS

. Can someone point me to the right direction?

Hi I am totally new to Prefect and trying to figure things out. I have successfully deployed Prefect Server using helm with my own ingress host. Now I am trying to register flow in my own notebook and wondering how to connect to my own server. Can someone guide me?

Hey all! I am looking to add persistence to the prefect server, I went through the documentation and found out information for using a local file mount volume or postgreSQL, However I wanted to know if prefect server supports mysql as a backend DB for persistence, If yes I would appreciate it if somebody could point me to the documentation for it. Thanks!

Hi, Can anyone point me towards the right direction on how I can setup access for the Kubernetes Agent to read/write Results to a S3 bucket using a AWS iam roles? I have used Kube deployment and RBAC to setup the agent and I am not using a service account atm.

Hi Why does this GQL query return "empty results"? I expect there to be only one, since the query filters the flow runs by id?

query {
  flow(where: { name: { _ilike: "Energy Insight" } }) {
    flow_runs(where: {name: {_eq: "stalwart-ostrich"}}) {
      name
      id
    } 
  }

->

{
  "data": {
    "flow": [
      {
        "flow_runs": []
      },
      {
        "flow_runs": []
      },
      {
        "flow_runs": []
      },
      {
        "flow_runs": []
      },
      {
        "flow_runs": []
      },
      {
        "flow_runs": [
          {
            "name": "stalwart-ostrich",
            "id": "8bca3ab2-05cc-4223-93f0-071813528545"
          }
        ]
      }
    ]
  }
}

Hey everyone I have a flows that triggers other child flows and I want to notify slack when one of them fails- this is my ( logic) code:

child_flow = create_flow_run(
            flow_name=CHILD_FLOW,
            parameters={
                "input_data": input_data,
                "run_id": run_id,
            },
            project_name="default",
        )
child_result = get_task_run_result(
            child_flow,
            task_slug="child_flow-copy",
            poll_time=3,
        )

I tried to add handler to get_task_run_result but it get error:

child_result = get_task_run_result(
            hazarder_flow,
            task_slug="child_flow-copy",
            poll_time=3,
            state_handlers=[post_to_slack_task_handler],
        )

TypeError: got an unexpected keyword argument 'state_handler'

So how can I do it? thanks

Hello fellows, i want to create child flows. For example i have a task that returns list. I want to create child flow to handle these every single elements. Can i do that?

I should have n+1 flows.

Hi all! I am running a Shell Task to kill zombie processes generated by one of my other tasks. The flow was functioning properly until 2 weeks ago but I cannot figure why it stopped working. Here is the associated flow: kill_geckodriver_task = ShellTask( log_stderr=True, return_all=True, stream_output=True) with Flow('kill_geckodriver', schedule=schedules_prefect[ 'kill_geckodriver']) as kill_geckodriver: kill_geckodriver_task(command='source {}'.format(os.path.join( os.environ.get('BASH_SCRIPTS_FOLDER'), 'kill_geckodriver.sh'))) The bash script is below: __ pkill geckodriver pkill firefox I can run the flow when the bash script only echoes a string, so the bug is not due to the flow or the path passed with an environment variable. I guess the problem lies in the sudo rights needed to run the "pkill" command. I have been trying to replace the current script with the following lines (replacing password with the user password), but with no success so far: __ # export HISTIGNORE='*sudo -S*' # to be added in production to avoid logging passwords echo "<password>" | sudo - S pkill geckodriver echo "<password>" | sudo -S pkill firefox Unfortunately, the flow still raises an error, and I cannot figure why. I have been trying to log it with the "log_stderr=True, return_all=True, stream_output=True" kwargs of the ShellTask but the only logs I have are joined as a screenshot. Any thoughts about it? The problem is probably password-linked, but I cannot seem to find an appropriate solution. Thanks a lot in advance!

Greetings Everyone, I'm attempting to use

prefect.Client

to query the GraphQL endpoint, however, I am getting errors that have me stumped. The relevant code:

import prefect
import pandas as pd

client = prefect.Client()

client.graphql("""
    query {
        agent{
          flow_runs(limit:5, where:{state:{_eq: "Success"}start_time:{_gt:"2022-02-14"}}){
            start_time
            end_time
            state
            flow{
              name
              id
              tasks_aggregate{
                aggregate{
                  count
                }
              }
            }
      		
          }
          
        }

    } """)

The errors:

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 465, in _request
    json_resp = response.json()
  File "/usr/local/lib/python3.8/site-packages/requests/models.py", line 898, in json
    return complexjson.loads(self.text, **kwargs)
  File "/usr/local/lib/python3.8/json/__init__.py", line 357, in loads
    return _default_decoder.decode(s)
  File "/usr/local/lib/python3.8/json/decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/local/lib/python3.8/json/decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/runpy.py", line 194, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "/usr/local/lib/python3.8/runpy.py", line 87, in _run_code
    exec(code, run_globals)
  File "/workspaces/prefect/ETL/CDNY/GraphQL.py", line 7, in <module>
    client.graphql("""
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 298, in graphql
    result = <http://self.post|self.post>(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 213, in post
    response = self._request(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 468, in _request
    raise ClientError(
prefect.utilities.exceptions.ClientError: Malformed response received from Cloud - please ensure that you have an API token properly configured.

We're using Prefect core as our backend at the moment and the documentation for this is sparse when I google. Thanks

I know for an ECS task, it is possible to specify the amount of RAM/CPU that is available to each task. Would it be possible to set these same constraints on individual tasks?

Hi Community - I am using

croniter

to calc cron schedules inside of the prefect flow and registering on the cloud instance. When I kick of my flow I want to understand what is the last cron time schedule that was ran. These cron values I have placed them into a dictionary that has these times for a handful of different jobs. Then, I was to use this as a parameters to some code I am running. For Example:

croniter(SchemaNames[key], datetime.now()).get_prev(datetime).strftime("%m/%d/%Y %H:%M:%S")

For whatever reason the datetime.now() function is recalling the datetime that might be corresponding to the datetime when the flow was actually registered and not actually the runtime datetime.now(). Any thoughts how to achieve the results I am looking or guidance from this point? Much appreciated. p.s. I would assume I could use the mquery api and extract the last schedule start times for these parameters but, I thought this approach was easier.

Is it expected that a

Result

using a specific serializer (e.g.

PandasSerializer

) would use the same serializer when loading the result using

prefect.tasks.prefect.get_task_run_result

? I have a task that uses the following task decorator:

@task(slug="output", result=GCSResult("<path>", serializer=PandasSerializer(file_type="parquet")), checkpoint=True)

When I try to get the result by doing:

result = get_task_run_result.run(flow_id, "output-copy", poll_time=5)

I get an error that seems to indicate it's trying to use

cloudpickle

instead of Pandas:

File "/usr/local/lib/python3.8/dist-packages/prefect/engine/serializers.py", line 86, in deserialize
    return cloudpickle.loads(value)
_pickle.UnpicklingError: A load persistent id instruction was encountered,
but no persistent_load function was specified.

Is there a way to have

get_task_run_result

use a specific serializer?

Hello, Do all mapped tasks have to be completed before dependant mapped tasks can be run. Let's say you have a flow that has these tasks A -> B-> C. Tasks B and C are mapped tasks. Is it possible for B to execute right after an individual run of A? Before starting another execution of A Say Task A adds 10 to input and print the value Task B Adds 100 to the input and prints the value. the expected output for the input of task A [1,2,3] would be: 11 111 12 112 13 113 I know that the actual output could be in any order.

Hello, a dumb question, I am aware that prefect Parameters are used to define flow inputs. What would happen if I use python variables that were defined out of the flow instead of define Parameters and adding the variables in flow.run?

Hey 🙂 I tried to implement a subflow in my main flow (like this) :

@task
def subtask():
   return 1

with Flow("subflow") as subflow:
   subtask()

@task
def main_task():
   subflow.run()

with Flow("main_flow") as main_flow:
   main_task()

But I faced this Error : _Unexpected error while running flow: KeyError('Task slug init_dirs-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.\n- Did you change the flow without re-registering it?\n- Did you register the flow without updating it in your storage location (if applicable)?')_ I precise that I had already register the _main_flow_ (without the subflow inside) and it worked, but then I updated it and I registered the subflow, and it failed, could you help me please ? 🙏 The task "_init_dirs_" is in the main_flow, finishes in a success state, but this error append when subflow.run() is called (and init_dirs is not in the subflow)

Hi 🙂, I am scoring the documentation but haven’t found the answer yet: on prefect-cloud (in particular) is it possible to run a flow when it receives an http request? What would this look like in prefect?

Hi, I am trying to run a flow given its name and project. Since the client can only do this given an ID I was trying to find the ID via graphql. However I fail to construct a proper query to only find the active version of the flow.

{
  flow_group(
    where: {flows: {name: {_ilike: "%name%"}, archived: {_eq:false}}}  ) {
    id
    name
    flows{
      name
      id
      archived
    }
  }
}

Will not actually filter for not archived versions. What am I doing wrong?

Hi there, I am trying to get more information about how a Kubernetes Agent in prefect cloud is configured. The setup was done by a colleague and is working fine, but I have a hard time to understand which clusters/workloads within the Kubernetes Engine on GCP are actually used by the prefect agent. The config tab in the web UI is rather empty 🙂 Thanks for your help!

Hi, we have a Parameter that is a date type. This gives us a date picker in the UI, but it is setting a datetime instead of a date (see screenshot).

Can a flow with multi tasks be deployed as multi Kubernetes jobs, because each task has a different docker image?