Hi, I have a pretty simple setup and flow (works fine locally), but I cannot make it run when I choose storage configuration to be S3 and run from the cloud with ECS agent and executor. I don't do anything unusual - simple flow with a few simple tasks. The error I'm getting is

Failed to load and execute Flow's environment: FlowStorageError("An error occurred while unpickling the flow: ModuleNotFoundError("No module named 'transform'")")

If I comment out transform, it will complain about next module. Project setup example is in the thread

Is it possible to fail a flow, if the agent is not available at the scheduled time? I would prefer this behavior in my case instead of the flow being marked late and start when the agent comes online.

Hey all, it was my understanding that if any task in a flow run fails, the flow will end up as failed, however I've just had my flow return with state Success despite having 1 failed task run, and 3 skipped task runs. Is this expected behaviour? Context - it's a very large static flow ~1250nodes. Prefect version 0.14.14, K8s + Dask execution

Hi! Yet another logging question.. I am trying to add a file logging handler that captures everything, while the “prefect” logger only shows INFO and up. I have tried searching around in the website, github and here, but with no luck. Right now I have the following MVE (see thread) and I am running prefect locally. Everything I do changes both the level of the prefect logger and the file-logger at the same time. How do I unentangle them? Also, it would be great if it could work with a local dask executor!

Hello all Is there any way to add a simple help text (or possibly even HTML) with the parameters? My team would be overjoyed... I'm posting this specifically here because it would be easiest to have an argument directly to prefect.Parameter(). Which is then displayed on the server (and could even used on the command line). Thanks!

Hello everyone, I ‘m trying to run my flow using AWS ECS Fargate (following the great step-by-step of Anna Geller: https://towardsdatascience.com/serverless-data-pipelines-made-easy-with-prefect-and-aws-ecs-fargate-7e25bacb450c#3e93). When running the flow, I’m getting an unplickling error

Failed to load and execute Flow's environment: UnpicklingError("invalid load key, '{'."

I’ve read the Slack thread that referred this error: https://prefect-community.slack.com/archives/CL09KU1K7/p1623777537484700?thread_ts=1623704787.418000&cid=CL09KU1K7 which seems to point to version compatibility issues. My setup : • Agent Prefect version 0.15.0 • Running on Prefect Cloud (“core_version”: “0.14.22+9.g61192a3ee”) • My task is a simple hello-world log, deployed with

with Flow("s3_flow", storage=S3_STORAGE, run_config=ECSRun_CONFIG) as flow:

•

is_serializable(flow) = True

Should I be downgrading my agent to match the server version ? Any help will be very welcome !

If I change the executor for an existing flow and reregister it on Prefect Cloud, shouldn’t that trigger a new version to be registered? Or does the executor somehow get updated without a new version? It’s hard to tell what’s going on because the Cloud UI, as far as I can tell, doesn’t show you what executor is configured for a given flow. So to be safe I’ve run

prefect register --force

to make sure the new executor config is being sent up to Prefect Cloud. But I don’t know if that’s necessary.

I'm running a flow that uses two subflows via

StartFlowRun

and the parent flow is throwing

ValueError: Failed to find current tenant None in result {'data': {'tenant'...

The subflows are triggering and running correctly, but it looks as though the parent flow can't view their status? Does the agent running the parent flow need to have a user API key or something attached to it in order to query flow status, but not to submit a flow for execution?

I was playing around with Prefect 0.15.0 and I stumbled upon the

prefect build

command, which seems like an interesting command for my use-case. However, I cannot find any documentation on this command. Is there any, and if so, where can I find it? Thanks!

Hey is there a graphql call that can be make to toggle off the schedule for a flow?

If another user is developing in the same flow repository, do they have to start their own agent? We're running the cloud backend.

Random: Prefect according to copilot 🙂 -- It gets more interesting but I thought this was neat. ....

Hi there! 👋 I'm using Prefect v0.15.0 and following the docs to set up slack notifications and I'm getting an odd error:

Exception raised while calling state handlers: ValueError("Failed to find current tenant None in result {'data': {'tenant': [{'slug': 'myTenant', 'id': myTenantId}]}}")

Any ideas to what could be causing this?

Hello! If I have a flow that is successfully check-pointing results (with S3Result in my case), what's the best way to have the flow use the checkpoint in successive invocations as a cache?

Hi everyone — I’m trying to use the compiled “task run name” on my custom state handler, but I’m struggling to find it on the “obj”. Does anyone know if that’s possible?

Hello folks! I'm just here to browse and carouse, a I'm still learning how prefect works. I'll be mostly a passive user, reading others answers and silenty judging everyone, but I'll get in touch if the need arises. Pack a towel!

If I pass a

Parameter

directly to a

DatabricksRunNow

task, Prefect detects the dependency and passes the information from the parameter to the task correctly. e.g.

path = Parameter("path")

DatabricksRunNow(...)(
    databricks_conn_secret=SECRET,
    notebook_params: {
        "path": path,
    }.
)

However, if I plug the parameter into a formatted string, for example, the information is no longer passed from the parameter to the task correctly.:

path = Parameter("path")

DatabricksRunNow(...)(
    databricks_conn_secret=SECRET,
    notebook_params: {
        "path": f"{path}",
    }.
)

In this case, the parameter class instance is plugged into the string vs. the actual parameter value that we want. So the notebook is given a

repr()

of a

Parameter

class instance — which is unusable, of course — instead of the string value of the parameter that we actually want. Why is that, and is there a way around this?

For a system executing flows in AWS ECR and orchestrated by Prefect Cloud, is there a way to redirect Prefect Cloud logs to CloudWatch?

Hi all ! I'm having a bit of trouble with documenting my project with sphinx or pdoc. For sphinx : I found this archived conversion which addresses the question of how to make a task's docstring available to sphinx, but I'm having an issue elsewhere : if I manually add a

..autofunction:: my_beautiful_task

directive, sphinx will document it according to its docstring without any issue, but if I rely on

..automodule::

to automatically document all the members (functions, classes...) of my module, it only finds regular functions and classes, not prefect tasks and flows. pdoc : I just gave it a quick shot, the same happends, it documents functions and classes, but not prefect tasks and flows :( Any idea? Thx !

We want to use AWS ECS to run flows using AWS ElasticFileSystem (EFS) to store the data created during data processing. We are able to mount the EFS from EC2 instance, but we are failing terribly when trying to do so when using runner

prefect.run_configs.ECSRun

. Our flow keeps starting and never completes - not even printing single line of text. Is there any working example of AWS ECS based prefect runner storing data on mounted AWS EFS volume?

I start local agent by open Power shell and "prefect start agent" (windows server 2012). But i want to run agent like services and auto start if server restart. P/s: My server cant install docker because docker not support Win Server 2012.

Hi everyone! I wanted to understand if there is a way to execute two tasks in the same executor in order to reduce data transfer. Is this something that is taken into consideration internally while building the dag or is their some way to let the executor know that this is a requirement. Any help would be appreciated. Thanks in advance!

hello everyone, I'm new to prefect, just started hacking today. I'm trying to use it to create dependency for my databricks jobs. But I can't seem to figure out how to create dependency between jobs. and I keep getting this error. (is imperative API the only way?). I'm running it with an agent in a VM if that's relevant

i have the following setup, where

flows.py

has my flow definition, and

utils.py

hosts some number of helper functions. And I'm using Github storage for the flow. Since Github storage only specify the path of the

flows.py

, it got module not found during execution. I wonder if Github storage does not support module? I can't seem to find a good descrption in the documentation. Thanks in advance if anyone can give me a pointer.

Hello everyone. I am Helm based deployment of Prefect. Setup is done with k8s agent for job propagation. I am confused for about how to go and load the files inside the platform. I see it supports Gitlab storage. But is there a way to sync the code to Prefect with CI?

I have a fairly basic question… Let’s assume I have some Python app that uploads something to S3. Now I want to kick off my flow. So I add this to my aforementioned Python app:

from prefect import Client

client = Client()
client.create_flow_run(flow_id="<some flow_id>")

But lo and behold, someone comes along an re-registers the flow with Cloud. The previous flow_id gets archived and a new one is generated. Now I have to go and update my Python app with the new flow_id? Can this be done by name or something? Or can I pull the flow_id as a variable? Or? Thank you.

Hello, trying to find a better way to use parameters for control flow using a dictionary (code in thread)

Hello i am working with a simple ml model and trying to use prefect ui to see the flow of it. I tried almost everything to debug it following the old threads related to pickling error, but nothing works. Can anyone please tell what can be done. Error - Unexpected error: TypeError("cannot pickle 'weakref' object") ``````

Hi everyone. I started using prefect a couple of months ago and it's been a good ride so far. By default, if the run() method succeeds for a task, Prefect sets the state to Success and records any returned data. I would however like to disable this recording of returned data because of how much memory it uses up. The task i run returns data of about 130Mb and I have thousands of raw data to map using this task. I couldn't find information about this online. How do you suggest i proceed? I should also mention that I'm running the flow without prefect server. Thank you for your time.

is there a way to report specific items on the slack hook automation? Like if I wanted to report "max date downloaded" for example. I have a special notification that will do this on fail signals, but part of that process is to actually raise the signal and call signals.FAIL(). Wondering if I could just do the opposite for success? Or would it end up being two separate slack messages, one being the automated slack flow success message, and the other being the custom signals.SUCCESS() ?