prefect-community #prefect-community

Hi, I'm running local prefect server in docker, and to register my flow I run a docker client. I am using this method(https://github.com/flavienbwk/prefect-docker-compose). I can see the code run successifully in docker container client and flow is registered but when I rerun the flow from UI I getting no module Pandas found. I have Pandas installed on host and also in the client container, my question is where else should the module be installed for the flow to run successfully from the UI?

Rhys Mansal

04/26/2022, 9:22 AM

Hi all, I'm having difficulty with the following code.

Copy code

param_a = flow.add_task(Parameter("param_a", default=True))
with case(param_a, False):
    task_a = some_task(arg)
task_b = some_task(task_a)

some_task

is being run multiple times with different arguments. When

param_a

is false some of these should be skipped. If I do not use

@task(skip_on_upstream_skip=False)

in the task decorator on

some_task

then all downstream tasks from the first skipped task are skipped, whether they are inside a case or not. If I do, none of them are skipped regardless of what

param_a

is set to. Does anyone have any idea how to get only the tasks inside the with block to skip (and only when

param_a

is true)?

Florian Guily

04/26/2022, 10:20 AM

Hey guys, i have a quick question on good practice. How scalable is the mapping task ? are the children tasks "queued" if the map is pretty big ?

xyzz

04/26/2022, 11:13 AM

Copy code

Regarding storage the Orion docs say "you can also configure a self-hosted key-value store for testing.".
But what kind of key-value store is compatible and how to set it up for testing purposes?

xyzz

04/26/2022, 11:17 AM

I'd like to test orion cloud together with kubernetes running on my laptop via docker

xyzz

04/26/2022, 11:20 AM

would redis (https://hub.docker.com/r/ubuntu/redis) be compatible?

Malthe Karbo

04/26/2022, 12:41 PM

Hi, when trying to create a deployment to orion cloud, I get "422 unknown" exceptions being thrown

Nikhil Joseph

04/26/2022, 1:15 PM

Is it possible to stop a task map if any of the calls fail?

Marwan Sarieddine

04/26/2022, 1:28 PM

Hi folks, we have run into this error message:

Copy code

{'_schema': 'Invalid data type: None'}

twice over the last week over our many flow runs. It seems other folks have encountered this due to a version mismatch between the agent and their execution environment. However that is not the case for us - additionally the same flow run will proceed to run successfully for future runs without any changes from our end. See more details in the thread

Leon Kozlowski

04/26/2022, 1:44 PM

Hi all - I’m seeing some flows being executed past their scheduled start time by roughly 15 minutes, when I check logs I see some Lazarus logs:

Copy code

Rescheduled by a Lazarus process. This is attempt 1.

What would be the best way to me to root cause this issue?

Andrey Tatarinov

04/26/2022, 2:01 PM

Question about the future of Prefect. I did not find references to Dask in Orion flow runners documentation. Does it mean that Dask cluster executor will be obsolete in the Orion?

✅ 1

Shuchita Tripathi

04/26/2022, 2:28 PM

Hi, I have a scenario where I need to modify the output of one task and pass it to the other. Is modification of task output and then passing possible? I am asking this because defining the tasks within flow does not execute them. So the output is not available. eg (is something like below possible): with Flow("flow") as f: abc = task1() # some modification required for 'abc' (return value of task1()). Modification like extracting a part of string from a longer string. modified_abc = # "string extraction from abc" def = task1(modified_abc)

Jason

04/26/2022, 2:29 PM

I'm having trouble locating the local_script_path:

Copy code

return os.stat(filename).st_size
      FileNotFoundError: [Errno 2] No such file or directory: 'hello-flow.py'

The S3 storage class is configured in a shared module between flows as such:

Copy code

storage = S3(
                bucket="EDITED-prod-platform-prefect",
                key=f"{project}/flows/{flow_name}.py",
                stored_as_script=True,
                local_script_path=f"projects/Examples/flows/{flow_name}.py",
            )

Bruno Murino

04/26/2022, 2:33 PM

Hi everyone — I’m trying to use the Prefect API to “clone” a flow and give it a different name, does anyone know if that’s possible at all ?

✅ 1

Milton

04/26/2022, 3:02 PM

hi how far back can one go in the audit logs?

Bruno Murino

04/26/2022, 4:09 PM

Hi everyone — are flow schedules preserved across flow versions ?

Lukáš Pravda

04/26/2022, 4:19 PM

Hello community, I’d like to test a rather simple scenario whether a task defined in a flow was run, but I dont seem to understand how to mock it properly. In simplicity my setup is similar to: file.py

Copy code

@task
def foo():
    ....

@task
def bar():
    ....

with Flow() as flow:
   try:
       foo()
   except:
       bar()

test.py

Copy code

from file import flow
from unittest.mock import patch

@patch("file1.bar")
@patch("file.foo")
def test_flow(mock1,  mock2):
    mock.side_effect = Exception("throw an error")

   flow.run()

   assert mock2.assert_called_once()

but the mock is never called, have found this:

<https://github.com/PrefectHQ/prefect/issues/1801>

, but could not really mount that solution to my exact problem. What am I missing? Thank you

Tom Manterfield

04/26/2022, 4:37 PM

On the Orion beta, I can’t see any route to adding GCP credentials to a

KubernetesFlowRunner

instance other than building the keys into the image itself. Has anyone else found a solution for this?

Xavier Babu

04/26/2022, 4:44 PM

We are not using any deployment container like Docker. We use the storage just mounted in the Linux Server. How can we create a local storage for prefect instead of using /tmp. When I run the the command "prefect storage create" I don't get any list to choose for storage type and whatever value I put, it says "invalid selection". Please help. Found the following storage types: Select a storage type to create: 4 Invalid selection 4

Kathryn Klarich

04/26/2022, 5:14 PM

Hi all, I’m not sure where the best place to post this is (here versus discourse), but I am running into an error when trying to register a prefect flow:

AttributeError: module 'lib' has no attribute 'X509_V_FLAG_CB_ISSUER_CHECK'

- it seems to be happening during this step

RUN pip install pip --upgrade

- has anyone come across this before and know how to fix it? I was able to successfully register this flow a few days ago and haven’t changed much in the requirements since then.

discourse 1

Bradley Hurley

04/26/2022, 5:21 PM

Hi Prefect Experts - Is it possible if I have a flow run thats in a failed/cancelled state, to clear the state of a task thats already succeeded so that when the flow is restarted the previously successful task will re-execute? I know you can do that sort of thing with Airflow, but wasn't able to replicate with Prefect.

Jai P

04/26/2022, 5:39 PM

hey all! question about the eventual introduction of the

case

statement in prefect 2.0: is there a rough timeline for when that may be introduced? Also, are there any major differences that are planned between how they work in prefect 1.0, where i think you can only conditionally go between tasks (to, say, possibly supporting subflows)?

Anna Geller

04/26/2022, 6:42 PM

Hi, Prefectionists! 👋 Want to chat about best practices in workflow orchestration? We've just created a new channel #best-practices-orchestration. Especially as we continue to iterate on Prefect 2.0 to be the best workflow orchestration engine everyone will love working with, we welcome a dialogue about best practices and your expectations! prefect2 cloud2 Also, if you want to chat about various trends in the data industry or share some tricks to work with data more efficiently, the new channel #data-tricks-and-tips is the best place to start a conversation. Lastly, to celebrate how marvin duck conquers the world, we started a channel #marvin-in-the-wild posting pictures from places Marvin has visited so far. Make sure to join the channels - we are excited about all data and workflow orchestration chats. Happy Engineering!

upvote 2

Slackbot

04/26/2022, 6:55 PM

This message was deleted.

✅ 1

Dylan

04/26/2022, 7:20 PM

Have a few questions on flow boilerplate and k8's templates

Michael Moscater

04/26/2022, 7:52 PM

Hello, is it possible to have a parent flow run a mapped create_flow_run task with a flow and multiple parameters, so the the created flows would be able to use the different parameters passed?

Josephine Douglas

04/26/2022, 11:33 PM

Hello! I am trying to create a flow that runs a task, sends the data resulting from that task to a secondary flow, waits for the secondary flow to run successfully, then runs another task. [code in thread] The primary and secondary flows are both registered in the same tenant and project but are set to use different images. When I try to run the primary flow, I get this error:

Failed to load and execute Flow's environment: TypeError("got an unexpected keyword argument 'raise_final_state'")

What am I doing incorrectly here?

discourse 1

Slackbot

04/27/2022, 1:32 AM

This message was deleted.

discourse 1

Raviraja Ganta

04/27/2022, 6:28 AM

I am new to Prefect and want to understand couple of things. • I want to run flows in GPU instance. This means that I have to run the prefect agent in GPU. Does this instance needs to be switched on all the time? How can we scale down/up based on the requests? • I want some other system to trigger the flows (backend systems in my org). For that I am thinking to write APIs for all the flows and trigger them when an API is called. If flows are invoked like this, will this still be shown in the UI? • How exactly prefect agent knows what code to run? Does prefect core code needs to be running all the time? • What are some best practices used for building applications using prefect? - I am getting confused regarding how code is managed Thank you for the help provided 🤗

Florian Guily

04/27/2022, 10:08 AM

Hello guys, quick questions about retries behaviour. I have a terminal mapped task that has retry policy. The retries seem to work well but i'm a bit worried about the logs. The delay between each retries is 10 secondes and for testing purpose, it is a tiny mapping (tried with 6 children tasks) and the flow is launched locally from terminal. So, between each retries, there is time where there is no task to execute so prefect enters in a "waiting" state according to the log (

Waiting for next available Task run at 2022-04-27T10:00:35.571259+00:00

). After a few seconds, a new log message states:

Beginning Flow run for 'flow_name'

and i have all of the previous log message of the previous tasks, as if the flow was re-executed. Is it normal ? Is the flow really reexecuted ? When executing this flow from prefect cloud, i don't see those logs so i'm a bit confused.