Hi all, what is the recommended way to deploy flows that come pre-built in 3rd party packages? Think e.g.

scikit-learn

providing a flow to train a machine learning model and you can deploy it with your own model class like

Deployment.build_from_flow(flow=sklearn.flows.training, …, parameters={"model": SomePythonClass})

Is that even feasible? I guess another way to look at it is “prefect collection but with flows, not tasks”

Can anyone explain why this never completes? I've also tried it with a dask runner and it also hangs. This is with Prefect 2

from prefect import flow, task


@task
def task1(x: int) -> int:
    return x + 10


@task
def task2(x: int) -> int:
    return -x


@flow()
def run_my_flow(n: int):
    task2.map(task1.map(range(n)))


if __name__ == "__main__":
    n = 500
    print(run_my_flow(n))

Hi again, I had a quick question regarding saving blocks with

is_anonymous

set to

True

. Is it important to also set the flag for

overwrite=True

if the anonymous is one that is saved numerous times ala a CI/CD process?

new here, is there a link to where i can read about SSO connections? I am trying to connect Prefect with our Okta instance.

Hi once again, I wanted to see if there is a pythonic way to configure the execution environment in our CI/CD process to submit to Prefect Cloud than what is described here: https://docs.prefect.io/ui/cloud-getting-started/#configure-execution-environment I know I can use os/subprocess calls to run the commands described here, but it would be preferred to use some conventions that might already exists. I'm searching around and haven't found anything yet, so it'd be great to get pointed in the right direction

Hello Everyone 🙂 I'm happy user of 1.0 and wanted to start migration to 2.0. I deployed locally orion server and started agent inside docker container. When I run flow from inside of container then I have success and flow run normally. But when I wan't to schedule or just run it from UI flow it's late, and nothing shows on logs in container. Addition info i successfully deployed and applied flow (prefect deployment apply my_favorite_function-deployment.yaml Successfully loaded 'log-simple' Deployment 'my-favorite-function/log-simple' successfully created with id 'f44fb717-a757-4603-aca8-df122b5fbf54'. View Deployment in UI: http://192.168.1.24:4200/deployment/f44fb717-a757-4603-aca8-df122b5fbf54 To execute flow runs from this deployment, start an agent that pulls work from the 'test' work queue: $ prefect agent start -q 'test') Starting v2.6.3 agent connected to http://192.168.1.24:4200/api... _ _ _ _ _ _ _____ _ _ _ _ _ _____ | _ \ _ \ __| __| __/ _| _| /_\ / __| _| \| | _| | _/ / _|| _|| | (_ | | / _ \ (_ | _|| .` | | | |_| |_|_\___|_| |___\___| |_| /_/ \_\___|___|_|\_| |_| Agent started! Looking for work from queue(s): test...

I’m migrating a project from Prefect 1 to Prefect 2. I’m getting

TypeError: task() got an unexpected keyword argument 'nout'

— is the

nout

concept longer required?

Trying to migrate to Prefect 2 using the flows defined in Docker infrastructure images (https://www.prefect.io/guide/blog/prefect-2-3-0-adds-support-for-flows-defined-in-docker-images-and-github/). In Prefect 1, we could tie the flows in each release to a specific Docker image version by setting the version in DockerRun(image="{image}:{version}"). That way we were certain that the flow were always running in the intended image. In Prefect 2: 1. Not setting the version in the infrastructure block, makes each flow use the 'latest' version. As far as I understand this means that a flow could potentially run on an older version if the images haven't been updated on the agent. In our old setup this would have cause a flow failure - which would be far better. 2. Setting the image version explicitly in the infrastructure block would mean that I have to create a new infrastructure block for every release we do. Am I missing any smarter ways to control versioning of flows and infrastructure?

Hi. I see that the issue I raised some time ago was fixed, but I still see the same behaviour in the Cloud. Is it fix only for scenario when you are running own backend?

Hello experts~~ I'm using prefect 1.0! I say thank you so much for always answering me quickly before asking a question.💙 When the flow, which was well executed for small-scale data, tried to receive a large amount of data from the our web server as parameter, the following client error was encountered. I downloaded the log of the flow from prefect cloud UI and attached it.(logs.json) and wirte here, too.

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': 'State payload is too large.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])
Traceback (most recent call last):
  File "/home/da/enviorments/bdi/lib/python3.10/site-packages/prefect/engine/cloud/task_runner.py", line 91, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/home/da/enviorments/bdi/lib/python3.10/site-packages/prefect/client/client.py", line 1604, in set_task_run_state
    result = self.graphql(
  File "/home/da/enviorments/bdi/lib/python3.10/site-packages/prefect/client/client.py", line 464, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': 'State payload is too large.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

Actually, it wasn't the first time that task failed, and before that, it ran for about 3 minutes and then informed me that no heartbeat detected. (

No heartbeat detected from the remote task; marking the run as failed.

) So, the solution I found was configure heartbeats to use threads instead of processes and worte about the flow run config in the .py file. When I did that, the task of receiving parameters was performed longer than the first... (1st try : 3 minutes running -> 2nd try: 12 minutes running) Although the task was executed for a longer time, it was still a failure. 😞 How can i solve this problem? What's the problem of my engineering? This flow may have been written in the wrong way, so I also wrote the flow code below..

import...
from prefect.run_configs import UniversalRun

def custom_function():
    '''some works'''
    return output 

@task
def parsing_waveforms(download):
    processing_target = download    
    '''some works by using above custom_function'''
    
with Flow('flow_waveforms')as flow:
    heir = Parameter('download')
    task1 = parsing_waveforms(download=heir)

flow.run_config = UniversalRun(env={'PREFECT__CLOUD__HEARTBEAT_MODE'}:'thread')
flow.register(project_name='data_factory')

I'm trying to post a message logs in slack after the flow gets to success. But it shows None as the message for some reason. I have used the same function for failure and it works fine on it. Is there any function for getting the set of data on slack for successfully completed prefect flows?

Hey all! I am using the latest Prefect version (2.6.3). The agent is deployed on AKS in the namespace called

dev-prefect-2

. Which makes no sense - why would I need privileges in the

default

namespace? I can run this with Prefect 2.3 without issues.

Hi all, when doing a flow run with Concurrent flow runner, my run always ends af 204 tasks with a timeouterror. I have tried setting the PREFECT_ORION_DATABASE_TIMEOUT to 10, but it results in the same error. Database being used is SQlite.

When defining an ECSTask if the provided task definition arn has a container named prefect (as prefect expects) the system will override the image in the task definition with the base prefect 2.0 one unless an image parameter is also supplier to the ECSTask. This is counterintuitive.

In Prefect v1, is there a way to set a label on a schedule through the UI? I see that the documentation mentions labels on clocks, the clock schemas in the serialization modules have label fields, but the UI and the GraphQL endpoint do not seem to recognize that clocks have labels. Am I missing something?

Hi All, can you please let me know where can i get the API url for prefect2 installation to run a test job remotely from my machine using curl

in Prefect 2

never mind...i got it

Hey guys! I have a Prefect Agent running as an ECS Service. It is uses a lightweight container and will be invoking deployments via

ECSTask

. My question: Do I need to install dependencies like

prefect-aws

,

s3fs

etc. and register the blocks (eg:

prefect block register -m prefect_aws.ecs

) in the container for the agent? Or can I just use the latest

prefecthq/prefect

image?

Marvin at DBT Coalesce this morning in #prefect-dbt prefect duckhttps://prefect-community.slack.com/archives/C0470L5UV2M/p1666274979508959

Am I missing something in the prefect REST API documentation? How do I authenticate with the API? https://docs.prefect.io/api-ref/rest-api/

In all my task runs, the orion ui shows Task Run ID to be identical to Flow Run ID. Am I missing something here or is that a bug? Prefect 2.6.3, local installation.

must

get_run_logger

be called in each task/flow? i have a flow with ~20 tasks. and after upgrading to Prefect 2, I have 20 different calls to

get_run_logger

— is there a better way to adhere to DRY?

Question about Deployments & UI, everything worked till version 2.2.* but after 2.3.* Deployments are not showed in UI. In same time - if check Networking - API answer with the list of deployments (see the screen in the thread)

G'day - prefect 1.0 question. If I have a "master flow" and I trigger a "subflow" with

create_flow_run

if that subflow then triggers more "subsubflows" if one of them fail, will my master flow know about it ? is there a way to propagate any of these signales?

Has anyone seen this errot with prefect cloud? I have no issues runnign the flows locally but when i trigger them in the cloud i get

result = self.result.write(value, **formatting_kwargs)
TypeError: LocalResult.write() got multiple values for argument 'self'

if after running a concurrent task run i want to run another task ( after completion of previous concurrent task run) with the output of 1st task in concurrent task run. Is it possible ?

When deploying through the CLI with --cron command.. Is there a way to add the timezone argument to this as there is when creating a deploy via python?

Hi - Is there any way I can upgrade my prefect 2 account to an Org easily and carry over the workspaces/users

Hey there! I have a really weird issue with Kubernetes Jobs that I can’t wrap my head around: I try to start around 150 flows from on orchestrator flow. I do this because I need to parallelize across K8s pods instead of threads. This works, but only around 50 of my Flow-Runs actually transition into

Running

state and start a K8s job & pod, the other 100 Flow Runs just idle around in “Pending”. I have searched everywhere to find hints about why they won’t start. There is definitely enough K8s computing resources for them to be scheduled, there is no concurrency limits set via tags or on the work queue directly. The K8s work queue just shows them as “Pending” in the UI. Does anyone have any idea where else to look? Cheers 🙂