hey folks. Hopefully simple question, I’m testing out prefect as a replacement to my org’s current tools. I want the ability to restart a failed flow manually, AFTER I make changes in the code base. Basically — I’m anticipating flows to fail and I’ll need to make tweaks to our credentials or logic in the code. But I don’t want to re-run the entire flow, just pick up at the current task where it failed after making changes. I can’t seem to register the new version of my edited flow as the same version id as the prior one. I’d need to do this because I need to restart the prior version that failed. The only thing I can do is publish the new version say (version id 15) and then version 14 that failed get’s archived and I can’t restart it with the new code base from version id 15

Hi all, I’m wondering where most people have the prefect agent running? Do you use a virtual machine or something like Azure Container Instance? Thanks!

Is prefect2 beta down for everyone or just for me?

When will secrets be available in the prefect cloud beta? Will the mechanisms be different from prefect 1.0? I'm not seeing anything in the docs that are specific to prefect 2.0. I do not have experience with prefect 1.0. I'm creating a prototype to see if our existing workflows can be ported over to prefect 2.0 when it's ready.

In the prefect 2b8:

File "/Users/****/Library/Caches/pypoetry/virtualenvs/equation-kGZ4A1K9-py3.8/lib/python3.8/site-packages/prefect/cli/cloud.py", line 231, in login
    exit_with_success(
  File "/Users/****/Library/Caches/pypoetry/virtualenvs/equation-kGZ4A1K9-py3.8/lib/python3.8/site-packages/prefect/cli/_utilities.py", line 36, in exit_with_success
    app.console.print(message, **kwargs)
AttributeError: 'PrefectTyper' object has no attribute 'console'

When trying to run (which worked in 2b7:

await login(
            key=os.environ.get("PREFECT_API_KEY"),
            workspace_handle=os.environ.get("PREFECT_WORKSPACE_HANDLE"),
        )

Greetings, I am having an issue registering a Flow with a simple ShellTask (Prefect/Cloud 1.0). The RunConfig for the flow is pointing to a custom image in ECR that contains the binary that we are attempting to run via the shell task (using the ECSRun type). The path to the binary in the shell command does not exist locally. When I attempt to register the flow, it fails with a “no such file or directory” error. Does the register cli command attempt to verify the shell task locally? How can I get around this when I only want it to reference the location in the container image?

Regarding the Recent changes in Prefect Orion, Blocks - Is blocks needed if our ecosystem is Kubernetes + ConfigMap for Non-secure values and Kubernetes + Vault/Opaque secrets for Secure values ?

Thank you to everyone here! Just letting you all know July 22 will be my last day at Prefect.

Hi Guys, we are currently relying on Prefect 1.2 for some critical production ETLs. We are starting to discuss whether we should consider moving to Prefect 2.0. However, I can't find anything in the documentation related to an Imperative API, Do we still have one in Prefect 2.0?

we have a schedule with two clocks: every friday, and the first of every month. Sometimes these overlap. Will prefect generate two events, or just one?

Can we schedule flows so a new run is created as soon as the previous one finishes?

Hello! The Orion beta is amazing. Thanks for putting together such an awesome package. Could I get some feedback on the client's use with filters? I'm not sure if there's a better way to achieve what I'm doing:

# Minimal example of my current use, where I need to build filter objects

from prefect.client import get_client
from prefect.orion.schemas.filters import FlowFilter, FlowRunFilter

async with get_client() as client:
    response = await client.read_flow_runs(
        flow_filter=FlowFilter(
            name={"any_": ["example-flow"]},
        ),
        flow_run_filter=FlowRunFilter(
            state={"name": {"any_": ["Completed", "Running"]}}
        ),
    )

Is there a way to query with the client like this instead:

# Avoids the need to import and manually initialize FlowFilter+FlowRunFilter

from prefect.client import get_client

async with get_client() as client:
    response = await client.read_flow_runs(
        flow_filter={"name": {"any_": ["example-flow"]}},
        flow_run_filter={"state": {"any_": ["Completed", "Running"]}},
    )

Or even:

# Allows exact matching without use of {"any_": [...]}

from prefect.client import get_client

async with get_client() as client:
    response = await client.read_flow_runs(
        flow_filter={"name": "example-flow"},
        flow_run_filter={"state": "Completed"},
    )

Hey again 🙋 We’re trying to test out the (ad-hoc) Dask cluster option for execution but we’re getting errors it and seems like we’re missing something (code in the thread)

Aaand -- me again 😳 with a related but completely different issue 🤷‍♂️ Question: when using a

DaskExecutor

(that relies on a dask

KubeCluster

) - are task Results handled in the flow, or is the Result handling delegated in the dask workers? I'm asking because when we swap from a

LocalExecutor

or a

LocalDaskExecutor

to a

DaskExecutor

- suddenly our S3Results (which all our tasks are configured with) seem to fail on

AccessDenied

errors (for

PutObject

attempts) So logically it seems like they are being run from somewhere else that doesn't have the proper permissions (whereas the k8s job running the flow itself, does) --- am i missing something?

Can you use orion on MacOS? I don't see any mention of macs in the installation docs and it looks like orion's CI only tests linux+windows. I'm asking because my test suite is passing on linux+windows but failing on mac -- just want to know if the issue is on my end or not.

Hey I am trying to run a flow using DockerDeployment using prefect 2.0 I have created a docker image of my flow environment and uploaded it on docker hub. I am getting this error

docker.errors.NotFound: 404 Client Error for <http+docker://localhost/v1.41/containers/ab511ceee07d320a6e458ba7a063b9cf92ba177e49bf84657e85b0f613ae6f4d/wait>: Not Found ("No such container: ab511ceee07d320a6e458ba7a063b9cf92ba177e49bf84657e85b0f613ae6f4d")

What am i missing?

Hello, regarding Prefect 2.0: just wondering... are blocks supposed to be the final answer for secrets or will there be further abstractions for supporting e.g. Hashicorp Vault?

Hi! Thanks for the huge Prefect 2.0b8! I have a question regarding Deployments. Since Deployments based on

DeploymentSpec

are deprecated as of Prefect 2.0b8 release, does that mean that defining Deployments in YAML is going to also be depracated in the future? Can I still base my work on YAML?

This message was deleted.

Hey, prefect 1 here, just noticed that i have a flow in a "cancelling" status for 3 days now (i just noticed it now with the week end...) it is running on an eks fargate node also for 3 days and i don't know how to terminate it. Should i just kill the pod ?

Hi !When i am registering same flow which is stored in Git from different system version gets updated, but in the same system it gets skipped. Why this happens, do we have option to check what changed which creates new version?

Hello, is there an option not to store versions in prefect at all? We have git version insread.

Hello, also few questions about retention and maintenance (non Cloud): 1) Which table in DB do you recommend to maintain, any purge plans? 2) Does any service in K8s allow to scale? I.e. we have agent, hasura, apollo, graphql, towel and ui each having single pod in k8s. What if we add 1 more graphql pod for example? 3) Can we control job run history retention period?

Hello everyone! I am building a flow where inference on a pytorch model is mapped on a series of images which are read from the disk. Currently, I am loading the model within a dedicated task and passing it as an argument to the downstream

map

function in an

unmapped

call. I am receiving an warning related to the size of the ml model that is being shared in the task graph (more details in the thread). Is there an issue if I am receiving that error message? And is there a better way to share that ML model across the tasks?

Hey guys, I'm totally new to prefect. I'm trying to use it, but just after pip install prefect in my conda environnement, I have a module import error.

Hi community, I seem to be completely misunderstanding the principle of result within flows. My python code below does not print "just a string" to my prompt. If I check the vars i'm able to disect it down to mijn_flow.result which gives me a uniterable object.

from prefect import task, flow
import time
@flow
def mijn_flow ():
    return("just a string")

print(mijn_flow())

The output is: 135451.255 | INFO | prefect.engine - Created flow run 'brainy-parakeet' for flow 'mijn-flow' 135451.271 | INFO | Flow run 'brainy-parakeet' - Using task runner 'ConcurrentTaskRunner' 135451.302 | WARNING | Flow run 'brainy-parakeet' - No default storage is configured on the server. Results from this flow run will be stored in a temporary directory in its runtime environment. 135451.443 | INFO | Flow run 'brainy-parakeet' - Finished in state Completed() Completed() Can you point me into the correct direction? I'm trying to access the return object of my flow (which is a string in this case).

Hello, I am trying to run flows on a Kubernetes private cluster using the Dask executor and GCS storage. I am coming from a GKE test cluster where i was able to run flows using this approach. The thing now is that the private cluster runs behind a corporate proxy. I've set up the HTTPS_PROXY env variable inside KubernetesRun job template and Prefect is able to download the flow metadata from GCS. The problem is that the same pod is creating the Dask cluster and it fails with this error:

RuntimeError(f"Cluster failed to start: {e}") from e RuntimeError: Cluster failed to start: 503, message='Service Unavailable', url=URL('<http://proxy-ip-here>:proxy-port-here')

Any ideas on how to fix this? I've tried adding a NO_PROXY env variable alongside HTTPS_PROXY but it doesn't work. I am using Prefect 1.2 Thanks!

with Flow("hello-world"): task1 = hello_task() task2 = test(task1) with Flow("Multiple Flow"): db_table=create_table() raw=get_complete_date() parsed=parse_complaint_data(raw) populated_table= store_complaints(parsed) populated_table.set_upstream(db_table) from prefect import Flow from prefect.tasks.prefect import StartFlowRun data_engineering_flow = StartFlowRun(flow_name="hello-world", project_name='ETL - Prefect', wait=True) data_science_flow = StartFlowRun(flow_name="Multiple Flow", project_name='ETL - Prefect', wait=True) with Flow("main-flow") as flow: result = data_science_flow(upstream_tasks=[data_engineering_flow]) flow.run() flow.register(project_name="ETL - Prefect") I am facing some issues in flow after executing the code we can able to see flow in UI but We couldn't see task results and also its taking lots of time for processing.. can anyone help me out this

Background Existing Infrastructure: • We do have full fledged Kubernetes clusters running on top of AWS/EKS. • We have stable Jenkins CI/CD pipelines to build/deploy Docker Images and Helm Charts • We support multiple environments separated at namespace level • Our observability is via NewRelic • Our Credential Stores are wired very well in Helm chart via custom charts+annotations • Data Warehouse is Snowflake and data sources are plethora of DBs and API services • ELT is carried out by Meltano + DBT or Python + Celery • Everything running on Kubernetes I am future user of a workflow engine, I am asking around for evaluations, I constructed this so far. I have to confess spent more time at

#dagster-*

channels compared to other communities and and it shows in the recommendations chart below under the light of our existing infra setup. But I wanted to add more “fairness” to the evaluation ratings and I’d highly appreciate some constructive feedback from the community on where I can improve this rating.

hey, i have a flow that is always register even when there is no change to the code. Any idea on why this is hapenning ?