Does anybody have a best practice for storing large data as a

Result

in 2.6+ ? I am thinking of passing data between tasks, persisted into remote storage as parquet, for example (e.g. using spark)

Has anyone got an example of python deployments with KubernetesJob using infra_overrides - when running the job it keeps telling me "instance of JsonPatch expected", not sure what to pass in, prefect 2.6.2

Hey guys! I'm programmatically creating all my deployments and am wondering what's a nice way to create them idempotently. Meaning, I run the

create_deployments

script multiple times and overwrite the deployments (similar to blocks). I'm currently using the

.build_from_flow

and

.apply

methods but I keep running into warnings

Hi Team, is there a good documentation regarding Installation of Prefect V2 on AWS ECS ?

Hi, probably straightforward question on Prefect 1.x, how do you force a flow to fail if a task fails?

Hi team! I'm building out a deployment flow for our CI/CD purposes for k8s and was looking to use the the

KubernetesJob.job_from_file

function to read a yaml file with our full k8s manifest. The function works, however I noticed that we have fields that get overwritten with default values in this call for

_shortcut_customizations

in

build_job

, specifically the

namespace

and

image

properties. I know that I can use an

infrastructure_override

when setting up the deployment, however there should be option to simply read a yaml file with those fields.

Hi, I'm having some issues with the async python Orion client and what looks to be a cloudpickle error. Code/screenshots in thread.

(prefect 2.0, deployment on cloud) I am running into a situation where all tasks are completed successfully but the flow is still running, what could be causing this?

Hey All - In the process of upgrading to prefect 2, so far really loving it. Having a few issues though. I am trying to build a custom image to run our flows. Starting with

prefecthq/prefect:2.6.1-python3.10-conda

as a base image. Couple of questions: • When installing a python environment in the container does it mater what it is called? I see the image has a

base

env and a

prefect

env. Should packages be installed in

prefect

env of can I create a new one? • If I create a new env should the Dockerfile include a line to activate the correct env.? • Is there a recipe for creating a custom image from a conda base image? I looked around and did find one, but maybe I missed it?

One more question: we are trying to run flows in a docker container locally for testing. I see in the docs that: • You must configure remote Storage. Local storage is not supported for Docker. Does this mean that we need to setup remote storage to test running the flows in the container? I currently have something like the following:

deploy_import_1 = Deployment.build_from_flow(
    flow=import_1,
    name="import1",
    work_queue_name="imports",
    schedule=IntervalSchedule(
        interval=timedelta(hours=1),
        anchor_date=datetime(2021, 1, 1, 2, 30, tzinfo=pytz.UTC)
    ),
    infrastructure=DockerContainer(
        image="path_to_image/import-prefect:python3.10",
        image_pull_policy="NEVER",
        auto_remove=False
    )
)

Do I need to include remote storage to test? I am a bit unclear on all the pieces of the puzzle still.

Wanted to highlight the #show-us-what-you-got channel with our revamped Prefect Recipes by @Serina and a KV/Store built by @Michael Adkins 🎉

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