prefect-community #prefect-community

Iván Sánchez

04/01/2022, 7:49 AM

Hello! I want to build a flow several components of which need to load a heavy model before they do their work. This model takes more than two minutes to load, therefore I'd like to do some sort of stateful component that can have the model loaded in memory and when data comes to it, it just has to do the .predict(). I've seen that such thing as “stateful components” do not exist, and I shouldn't rely on a component's state because. Is there any guideline or best-practice on how to achieve this? Maybe Prefect is not intended for this and the components can't be so heavy? Thank you

Domenico Di Gangi

04/01/2022, 8:49 AM

Hi all, I am trying to run the orion server on a machine and open the ui from a different machine on the same network. When I type in my browser http//{ip server machine}4200 the first page of orion-ui loads fine but it does not fetch any info on deployments or current runs. I am not a javascript expert but, it seems that it tries to fetch data from http://127.0.0.1:4200/api while it should probably use http//{ip server machine}4200 . Is there a way to see the ui from a different machine, i.e. not from the same machine where the orion server is running?

Andrei Aldescu

04/01/2022, 11:14 AM

How to access the Prefect CLI ? is this something that I download on a windows machine ?

Patrick Tan

04/01/2022, 12:00 PM

Hi, Flow A repeatedly (loop) run Flow B. I want run next instance of Flow B after current Flow B is completed. I ran below and 2 instances of the same flow run at same time, and flow-id is same. see screenshot.

Copy code

with Flow("parent-flow") as flow:

    for i in range(2):
        flow_a = create_flow_run(flow_name="livelots flow", project_name="LiveLots-ETL")
        wait_for_flow_a = wait_for_flow_run(flow_a, raise_final_state=True)
        print(i)

Matthew Seligson

04/01/2022, 12:14 PM

What’s the best practice for running a flow of flows locally?

Mia

04/01/2022, 3:09 PM

Hello, is there a way to see in the UI which code is being run like in Airflow?

Wei Mei

04/01/2022, 3:30 PM

Hi! In my flow I call my tasks like this. I am experiencing a problem where the snowflake_load is running before the upload_to_s3 task.

Copy code

today, dir = get_data(endpoints=api_endpoints)
    upload_to_s3(today, dir, endpoints=api_endpoints)
    snowflake_load(today, schema="statistics", endpoints=api_endpoints)

Donnchadh McAuliffe

04/01/2022, 3:51 PM

Hey, I've signed up to Orion Cloud to continue my investigation of Prefect for my engineering team. I deployed a simple flow and am just trying to run the flow via a simple API request:

Copy code

POST <https://api-beta.prefect.io/api/accounts/123/workspaces/456/deployments/789/create_flow_run>

PAYLOAD
{
    "state": {
        "type": "SCHEDULED",
        "message": "Quick run through UI"
    }
}

Headers {Authorization: Bearer api_key}

Getting a

SSL Error: Certificate has expired

. Any ideas?

Joshua Greenhalgh

04/01/2022, 4:00 PM

Hi wonder if someone could help me - I am trying to implement a combination of module storage with kubernetes job run config - my process is to build a python module with all my flows within a docker image - push this image to a registry and then specify this image as the one that the kubernetes job runs - now when I register these flows with cloud if the code in the module changes I would expect to see a version bump however I just get the following message on subsequent registrations;

Copy code

Collecting flows...
Processing 'flows':
  Building `Module` storage...
  Building 'first_flow'... Done
Writing output to 'flows.json'
========================== 1 built ==========================
Collecting flows...
Processing 'flows.json':
  Registering 'first_flow'... Skipped (metadata unchanged)
================== 0 registered, 1 skipped ==================

Amogh Kulkarni

04/01/2022, 9:49 PM

Hi Prefect team. Is there any documentation on what is the fundamental difference between Flow ID, Flow Group ID and Version Group Id? I could spot few differences but haven’t quite understood the need for having three Ids.

Ken Nguyen

04/02/2022, 9:05 AM

I currently have a flow that is set up like so:

Copy code

with Flow("data-quality-tracking-model-run-duration-flow", run_config=RUN_CONFIG, storage=STORAGE) as flow:

  dbt_run_flow_run_id = create_flow_run(
      flow_name="test-dbt-run-flow",
      project_name="data_quality_tracking"
  )


  flow_run = wait_for_flow_run(
      dbt_run_flow_run_id, raise_final_state=True, stream_logs=True
  )


  flow_logs = get_logs(
      dbt_run_flow_run_id,
      task_args={"name": "Getting logs", "trigger": all_finished},
      upstream_tasks=[flow_run],)

get_logs

is a function that takes in a

flow_run_id

, create a

FlowRunView

, then get the logs of that

FlowRunView

. My issue is that despite having both upstream_task defined AND a FlowRunView.get_latest() for the

get_logs

task, I’m still getting logs that are incomplete. Do you have any suggestions for why my

get_logs

function is prematurely retrieving logs of the child flow/retrieving incomplete logs?

R Zo

04/02/2022, 11:42 AM

Copy code

Hi prefect team,

I am trying to spin up a flow of flows, so I started a prefect server and agent with label "test". Below is a snippet of code that should run child_flow_1 which is a flow with multiple tasks. However child_flow_1 does not run while prefect quickly returns a few success logs and there is no error reported. What I am missing? I have run child_flow_1 without the use of flow of flows and it works fine. Following is a snippet of the log using Dask, but I have tried LocalDaskExecutor as well.

test_flows created
Flow URL: <http://localhost:8080/default/flow/cd6dc2bb-8f54-4f5a-b046-d90e5c853dbe>
 └── ID: e405d852-5c75-4001-815e-619687e592d3
 └── Project: test_flows
 └── Labels: ['mymachinexxx', 'test']
[2022-04-02 22:30:01+1100] INFO - prefect.FlowRunner | Beginning Flow run for 'test_flows'
[2022-04-02 22:30:01+1100] INFO - prefect.DaskExecutor | Connecting to an existing Dask cluster at <tcp://192.168.20.9:8786>
[2022-04-02 22:30:02+1100] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded


test_flows = make_three_flows()
for tfl in test_flows:
        tfl.register("test_flows")
        # tfl.executor = LocalDaskExecutor(scheduler="threads",num_workers=10)

with Flow("parent_flow", run_config=LocalRun(labels=["test"])) as parent_flow_complete:
        working_dir = Parameter("working_dir")  #
        create_flow_run(
            flow_name="child_flow_1",parameters={"working_dir": working_dir})

parent_flow_complete.executor = DaskExecutor(address="<tcp://localhost:8786>")
# parent_flow_complete.register("test_flows")
parent_flow_complete.run(parameters={"working_dir": working_dir},)

Jai P

04/02/2022, 5:06 PM

👋 hello folks! I suspect the answer to this will be "use prefect 2.0" but wanted to ask: my company is investigating prefect for the use case of storing "DAG-y" transformations elsewhere (perhaps, behind some service interface), and dynamically executing them at runtime. I think prefect 1.0 won't solve this since the

flow

needs to be defined up front, but correct me if i'm wrong. I'll try to add a few more details in the thread, and thanks in advance for your help!

Shiyu Gan

04/03/2022, 2:36 AM

Can someone verify they can open the case study docs on this page? https://www.prefect.io/why-prefect/case-studies/ I can only open the Jellyfish case study.

Shiyu Gan

04/03/2022, 2:36 AM

The ones below simply do not do anything when I click to open then.

Atul Anand

04/03/2022, 1:13 PM

Just want to from which API can I create a dependent task? , A, B,C,D like these are my task and D task need to be executed when A and B task fails and C task will be executed with the input results of A and B.

Jai P

04/03/2022, 6:44 PM

Hi there! Can you visualize a flow in prefect 2.0, similar to

flow.visualize()

in prefect 1.0 (not using radar)? If not, are there plans to add that functionality?

Bruno Nunes

04/03/2022, 6:54 PM

Hello, I went through the steps in this tutorial to deploy prefect orion 2.0b2 in k8s (if you provide a namespace other than default the service and deployment are created but the role and the rolebinding are only applied if namespace = default). When I apply the manifest the service and deployment seem to be done correctly but it doesn't create a work queue as described here. Furthermore, the logs show an unusual IP used for the api URL:

Configure Prefect to communicate with the server with:

prefect config set PREFECT_API_URL=<http://0.0.0.0:4200/api>

Check out the dashboard at <http://0.0.0.0:4200>

12:51:20.774 | WARNING | prefect.agent - No work queue found named 'kubernetes'

I've edited the service to be type LoadBalanced and used the external IP to set the PREFECT_API_URL

prefect config set PREFECT_API_URL=<http://xx.xx.xx.xx:4200/api>

I've updated the sqlalchemy as suggested in here and created a new work-queue called kubernetes. I've created a new storage pointing to my azure blob storage and created the new deployment and run it. Everything finishes without errors but I don't see any activity in my cluster nor nothing being created in the UI. Can you give me some guidance on what I might be missing?

Atul Anand

04/04/2022, 2:46 AM

Inside flow can we use for loop , if else and other programming stuff?

Shiyu Gan

04/04/2022, 5:56 AM

Is this answer about Dask Scheduler fault tolerance still up to date?

✅ 1

Jeff Kehler

04/04/2022, 6:18 AM

I've created a flow that triggers other flows using the

create_flow_run

function. I would like to use a Parameter to configure the top level flow but it seems parameters are only really meant to be passed into tasks. Is it possible to use the value from a Parameter within a Flow?

Konstantin

04/04/2022, 6:56 AM

Hi Prefect team, in the current one, I can't do anything with the prefect constantly hanging. I need help. First, stop all running tasks in the scheduler, I attach a screenshot below. Secondly, to deal with the problem, fix and prevent the recurrence of this situation. The prefect is located on local servers, in docker, not in a cloud service. Tasks are performed in separate projects. Four agents have been launched that interact with GitLab. The first problem is that the agent freezes while performing the task, the monitoring runs out of RAM. In the task manager, in the docker "top", there are many python *.py launches, where parent ID =-1, kernel, docker container

Jan Nitschke

04/04/2022, 9:52 AM

Hi Prefect, I want to run a flow on ECS and use GitHub as storage. My python code imports modules. The flow definition looks something like:

Copy code

from tasks import my_task
from prefect.storage import GitHub
from prefect import Flow
from prefect.run_configs import ECSRun


storage = GitHub(
    repo="repo",  # name of repo
    path="path/to/myflow.py",  # location of flow file in repo
    access_token_secret="GITHUB_ACCESS_KEY",  # name of personal access token secret
)

with Flow(name="foobar",
          run_config=ECSRun(),
          storage=storage) as flow:
    my_task()

The problem seems to be that the GitHub storage only clones the single file and not the entire project which causes my import to fail. (

ModuleNotFoundError("No module named 'tasks'")

) I've seen that there has been some discussion around this issue but it hasn't really helped me to solve the issue.... Is my only option to clone the repo into the custom image that I use for my ECS task? But that would mean that I would have to rebuild that image every time I change something to my underlying modules, right?

🤪 1

✅ 1

Andres

04/04/2022, 11:54 AM

Hi Everyone, I'm developing a handler that will be sending slack notification to our team on flow failures with detailed information. I have managed it so far and it runs flawlessly locally. The issue comes when running it on our server, the handler sends a default notification that just says

Some reference tasks failed.

I was investigating a bit and seem that the state contains the result of all tasks while running it locally (

state.result

) while on the server this is empty (i printed it using the logger) . Any idea on how to address this?

discourse 1

✅ 1

Atul Anand

04/04/2022, 12:37 PM

Used Map functionality, But at a given point of time, I only see hike in only one cpi. Used DaskExecutor with 8 worker and 4 thread.Can anyone tell what is the issue.

Tom Klein

04/04/2022, 1:16 PM

Hello - I have a Q about ad-hoc env vars when running a flow — is it possible to “save” them somehow so they can be re-used again? or do I have to refill them manually each time i wanna create an ad-hoc run?

Shuchita Tripathi

04/04/2022, 1:59 PM

Hi. poetry add prefect[azure] is giving error. And I am guessing this is the reason, my flows are not working. Without azure they were working fine. What can be done here?

Rajan Subramanian

04/04/2022, 2:08 PM

Hello all, i had my prefect orion cloud running for last 4 days with no hiccups. today i logged in and i see all the deployments disappeared from the cloud. Is there a time limit for these processes? @Anna Geller

✅ 1

Joshua Weber

04/04/2022, 2:09 PM

Hey again everyone, is anyone implementing prefect while initializing local agents inside kubernetes pods? When our kubernetes pods die off and restart (for whatever reason) the old agents continue to live in a “dead” state in prefect. Eventually the agents accumlate and slow the whole UI down and deteriorate the performance. Anyone else have this problem or know a way to manage the agents in code?

✅ 1

Rajan Subramanian

04/04/2022, 2:44 PM

Hello couple questions, 1. After deployment, if i make further changes to my file. How does this new change get incorporated in the cloud? do i need to do a

Copy code

prefect deployment create deployment_name

again for those new changes to take affect? 2) if above is true, then do i need to rerun the tasks again on the UI? 3) sometimes i inadverently press run twice and i have two running processes. Is there anyway to stop a process after it has been started? 4) when i delete the workspace, to start over, i notice when i type,

Copy code

ps aux | grep python | wc -l

the python processes are still running and i have to do a

Copy code

pkill python

to kill all the python processes. Is there any way that once a workspace is killed all the python processes are killed along with it?