prefect-community #prefect-community

Greetings, I’m aware that the Dask functionality in Prefect allows for running parallel flows/tasks across different agents. I’m curious if the K8s operator does the same? Ultimately what I am attempting to do is offload a resource intensive task to different agents. The this particular task was previously done using native pipeline and a for loop. I’ve refactored this logic (in python) to use a thread pool and removed the for loop. While this had great performance improvements the timing of the logic is still not efficient enough. I am now considering moving this logic to a prefect flow in the hopes of improving this parallelization even further across different agents which are setup as kubernetes pods.

✅ 1

Abhinav Ayalur

12/05/2022, 1:05 AM

Hey there, I was thinking of using Prefect to route messages throughout my platform and I was wondering if there was a way to "join" lists of outputs created by two separate tasks? So if I have task 1 that outputs a list of elements and task 2 that outputs a list of other elements, assuming they have the same length what is the best way to join them together for my third task?

✅ 1

Abhinav Ayalur

12/05/2022, 1:27 AM

Separate question, was wondering how GPU jobs were orchestrated? We have some large GPU tasks that we'd like to write in, and were wondering if there was a way to have an init function that instantiates a model so we don't have to instantiate it for each call of a task, and that we could maybe have "warm tasks" running that could pickup these jobs faster without having to initialize the model again?

Oliver E

12/05/2022, 4:25 AM

Hello, small question. Sorry if I missed this in the documentation, but is there a way to view what is in a work-queue? Im running an agent locally (2.7.0) with a --limit. Threading with context.

Ben Muller

12/05/2022, 6:15 AM

Is there such thing as task concurrency limits? I have 8 or so tasks all running with

.submit

with the DaskTaskRunner
with no cluster. It seems like about two tasks run at a time and then another two start, my assumption was that all 8 would start and Dask handles them and they finish when they are ready? I know this would be limited by the threads too, but the machine it is running on should have more than enough threads to handle those 8 tasks.

✅ 1

Deepanshu Aggarwal

12/05/2022, 6:56 AM

hi! if i trigger a flow run via deployment. is there a way to get the output of that flow run in return or even using any api ?

iñigo

12/05/2022, 9:07 AM

Hi, Is it possible to skip a subflow if it fails, I mean in a flow of flows config to do the next flow. thanks

Joël Luijmes

12/05/2022, 10:44 AM

Hi there 👋 I’m working on a flow which (in theory) spawns off a lot of tasks in flow runs (about 10-50k task runs each day). I’m sure Prefect can handle such volumes, however there is also a bit of overhead associated with this (database / disk): • log records for “Created task run”, “Submitted task run” “Execution … immediately” • task_run record • task_run_state record • flow_run record Right now I got about 380k task runs and 17k flow runs in my database, with already 20GB storage. Understandably this originates by passing data to each sub flow. I suppose my current setup is not ideal (well unless I want to pay a lot for storage). Regardless of that, is there a way such that Prefect (Orion), performs roll ups on these tables? Or should I just delete these records myself from the database.

✅ 1

Vadym Dytyniak

12/05/2022, 11:17 AM

Hi. In Prefect 1 we use

ENDRUN

exception to fail and do not retry the task, does it exist in Prefect 2?

👀 1

Timo

12/05/2022, 1:26 PM

Hi all, is there a parameter to define when a flow is considered as late? So when we start at deployment manually the flow always went into state "Late" which then triggers a notification but few seconds later its running. Solution for that would be to define "Late" after e.g. 30 sec. Thanks

🙌 1

Luca Schneider

12/05/2022, 2:02 PM

Hi all, I noticed a weird behaviour lately with kubernetes jobs. My pipelines are doing fine but the flow runs are ending up in “crashed” state. The only error message I could find is :

Copy code

State Message
Flow run infrastructure exited with non-zero status code -1.

✅ 1

Justin Trautmann

12/05/2022, 2:30 PM

Hello community, hello Prefect team, the flow cancellation feature in 2.7 is very useful and exactly what our internal users were requesting. however, our tasks submit long running jobs to external services that would continue to run even if the flow running infrastructure is terminated. in order to save costs, it would be great to have the possibility to handle sigterm appropriately and cancel any external jobs before actually terminating. i played around with python signal handlers and found that the way how prefect executes the flow in a thread doesn't allow setting custom signal handlers the conventional way. are there any best practices available on how to deal with this type of situation? any input on how to react to sigterm from within a flow is very much welcome. thanks

plus one 1

✅ 1

Matthew Ferry

12/05/2022, 3:06 PM

Hello! I'm working with prefect to orchestrate some dbt tasks, but I'm struggling with ECS deployment. I have one flow that I want to run for multiple customers, so I need data access to be single-tenant, but it doesn't seem efficient to run an agent for each customer. I was hoping to: • Run one agent as a long-running ECS task that listens to one queue • When a flow run is requested, that agent spins up a separate ECS task (with client-specific permissions) to run the entire flow • The ECS task runs a prefect flow that includes dbt shell command prefect tasks from prefect-dbt Could somebody help me think through how this can be done? Most of the ECS docs describe v1 not v2, and I only found this discussion in Discourse which said ECS wasn't really supported.

Vadym Dytyniak

12/05/2022, 3:06 PM

Hi. How to test the Prefect 2 task that uses

get_run_logger

? When I try to use task.fn it fails with no task or flow context.

✅ 1

Vadym Dytyniak

12/05/2022, 4:06 PM

Hello. We see issue with SequentialTaskRunner. Why second task is not waiting for first one?

Copy code

@task()
def copy_file(n: int):
    get_run_logger().info('Start file copying')
    time.sleep(n)
    get_run_logger().info('Completed file copying')


@flow(task_runner=SequentialTaskRunner())
def sample() -> None:
    copy_file.map([3, 4])


if __name__ == '__main__':
    sample()

Logs:

Copy code

18:05:29.909 | INFO    | Flow run 'devious-aardwolf' - Created task run 'copy_file-6cc8b0a3-0' for task 'copy_file'
18:05:29.910 | INFO    | Flow run 'devious-aardwolf' - Executing 'copy_file-6cc8b0a3-0' immediately...
18:05:29.916 | INFO    | Flow run 'devious-aardwolf' - Created task run 'copy_file-6cc8b0a3-1' for task 'copy_file'
18:05:29.917 | INFO    | Flow run 'devious-aardwolf' - Executing 'copy_file-6cc8b0a3-1' immediately...
18:05:30.468 | INFO    | Task run 'copy_file-6cc8b0a3-1' - Start file copying
18:05:30.534 | INFO    | Task run 'copy_file-6cc8b0a3-0' - Start file copying
18:05:33.536 | INFO    | Task run 'copy_file-6cc8b0a3-0' - Completed file copying
18:05:33.726 | INFO    | Task run 'copy_file-6cc8b0a3-0' - Finished in state Completed()
18:05:34.469 | INFO    | Task run 'copy_file-6cc8b0a3-1' - Completed file copying
18:05:34.658 | INFO    | Task run 'copy_file-6cc8b0a3-1' - Finished in state Completed()

✅ 1

Tim-Oliver

12/05/2022, 4:37 PM

Hello everyone, How can I get the task-runs of a flow-run via REST API. I tried

tr = await client.read_task_runs(limit=1, flow_run_filter=FlowRunFilter.construct(id="562d1705cfcc48a0b55df98373fd10bd"))

Christian Vogel

12/05/2022, 5:14 PM

hi prefect community! we are using prefect notifications in the prefect cloud to detect late flows in order to check if our agents are healthy. In case a run is late we send a notification via slack to be informed. Unfortunately we encountered some false positives, so flows were marked as LATE but the agent is healthy. We saw that there is the attribute PREFECT_ORION_SERVICES_LATE_RUNS_AFTER_SECONDS, but this is only configurable on the self-hosted cloud. Any idea what we could do?

plus one 4

Ashley Felber

12/05/2022, 5:39 PM

Hello, I want to create an event driven (when files land in s3) deployment. Is this best way to do this with a lambda function? If so, what's the best documentation to follow on this?

Jean-Michel Provencher

12/05/2022, 6:48 PM

Quick question, do you plan on allowing to an additional properties when creating a deployment in Prefect 2 that would target a directory for the upload. Currently this seems quite limiting to only have the possibility to upload the current folder and ignore files based on the .prefectignore file. I see a dirty workaround of people actually uploading the whole directory on the first deployment creation, and then ignoring uploading for following deployments command. Ideally I would like to have the option to target specific files/folder for the upload.

Simon Macklin

12/05/2022, 6:50 PM

hey prefect we had some strange behaviour with our on prem prefect v1 stack, one hosted in aws and another in azure where flows will not run and we also had an unusual amount of Postgres activity, these are completely isolated from each other too would any issues with prefect cloud cause potential problems with the community v1 on prem solution?

1️⃣ 1

✅ 1

Mokshith Voodarla

12/05/2022, 7:13 PM

Hi! I was thinking of using Prefect but I'm not sure if it works well for my task. I am running large pools of GPU workers nodes which are orchestrated via some separate set of logic, and latency matters a lot. My thought was to set up Kafka as a messaging queue that all these workers listen too since it also offers the ability to do kSQL joins which are beneficial for me to say match up frames and frame metadata that might've been processed by different models upstream. I was also concerned about somehow having a warm pool of workers so the models don't need to cold start every time. I wouldn't even think about a workflow orchestrator but I want to have the benefits of task observability, logging, etc which it feels I wouldn't get if I used Kafka by itself.

Chanaka De Silva

12/05/2022, 8:16 PM

All, I am running a flow using prefect UI and hitting a

ModleNotFound

error. I register the code as follows

Copy code

prefect register --project CHX -p sparsify.py -n "sparsify" --label chx-2022-2.2

The folder where

sparsify.py

sits has the missing module’s python code; masks.py. And I import the module and relevant functions as follows

Copy code

from masks import MaskClient, combine_masks

Somehow when I run sparsify via UI it can’t find it. The code just runs fine in an

ipython

session

✅ 1

Jean-Michel Provencher

12/05/2022, 8:22 PM

Hello, the doc mentions that the .prefectignore file supports the git ignore patterns. However when testing I see that this does not support wildcard like "*" exclude everything. Am I the first to face this ?

Kristian Andersen Hole

12/05/2022, 9:49 PM

Hello, Prefect community! I have a question. I’m considering using ECSTasks for my execution layer setup, but wanted to verify something first. From www.prefect.io/guide...#NewDockerdeploymentexperience

If your deployment defines
-ib docker-container/prod
and doesn’t specify any remote storage block, Prefect will assume that the flow code is baked into the image.

Does this hold true for

-ib ecs-task/prod

as well?

aws 1

Neha Chopra

12/05/2022, 10:59 PM

Hi Everyone! I am looking for some advice on Prefect-DBT Core integration. My company uses Prefect version 1. I initially started with cloning DBT repo into Prefect method(referred online resource) which worked fine locally but not inside the Prefect cloud as our internal org repos have strict security policies that don't allow them to be cloned inside 3rd party products. Hence, I decided to try out the git submodule method where the DBT repo resides as a submodule inside Prefect repo and I use DBTShellTask to execute the DAG commands(like below). The flow works fine in my local but in Prefect, it fails due to statement helper_script=f"cd { DBT_DIR}" as it is unable to access the right folder. It evaluates helper_script as (cd: /Users/my-username/Development/prefect-repo/containers/dbt-prefect-container/dbt-repo)

Copy code

dbt = DbtShellTask(
        return_all=True,
        profile_name="dbt_repo",
        environment="dev",
        # profiles_dir=".",
        overwrite_profiles=True,
        helper_script=f"cd {DBT_DIR}",
        log_stdout=True,
        log_stderr=True,
        dbt_kwargs={ # removed for readability}
        )(command="dbt docs generate")

The folder structure is like: prefect-repo |- containers |- dbt-prefect-container |-dbt-repo(submodule) |- flows |-flow.py (from where I want to access the location DBT_DIR) My questions are: 1. Is it okay to use git submodule method inside Prefect? 2. How to pass correct repo location so that Prefect cloud doesn't start the root at local dir: /Users/my-username

1️⃣ 1

John

12/06/2022, 12:44 AM

How to set the time threshold for a flow run to be considered "Late"? (e.g. whether to wait 30 seconds or 30 minutes before marking a flow run as late) Related previous posts (none of those was ever answered): • https://prefect-community.slack.com/archives/CL09KU1K7/p1670260486146599 • https://prefect-community.slack.com/archives/CL09KU1K7/p1670246788470149 • https://prefect-community.slack.com/archives/CL09KU1K7/p1668495473066159

✅ 1

Michał Augoff

12/06/2022, 1:38 AM

hello! is it possible to build/use custom infrastructure types (beyond what’s available in the

prefect.infrastructure

module)? I assume no since this needs to be present in the agent’s env but double checking

✅ 1

Ton Steijvers

12/06/2022, 8:39 AM

Hi, In Prefect 1 we often see flows stuck in Running state because the underlying k8s pod might have been terminated. Prefect is for some reason not able to detect this and the flow stays in a running state. In one case the logs show

Flow run RUNNING: terminal tasks are incomplete.

as the last log message, then flow stays in running state forever. Is this a known issue?

1️⃣ 1

✅ 1

Sunjay

12/06/2022, 9:00 AM

Hi Team, How do i set infra_override in the deplyments yaml to override my ECS task memory for a particular deployment. Can someone please help me with this?

✅ 1

James Zhang

12/06/2022, 11:56 AM

hey guys, I’m trying to implement an exponential backoff retry in a @task where I use native python http Retry class, however in the execution of the task I don’t see the retry being invoked in the Flow? can someone enlighten me on this one?