I have a task that I think returns only 1 result, but I keep getting the below error. Is there a reason why this task thinks it’s returning multiple results? (code in thread)

TypeError: Task is not iterable. If your task returns multiple results, pass `nout` to the task decorator/constructor, or provide a `Tuple` return-type annotation to your task.

I have a Prefect flow that fails when it is unpickled, the error message is related to me decrypting variables in pydantic's Settings().. is there any good way to move forward and debug pickling issues like this?

running on ECS (trying to), how can I set the equivalent of

prefect backend server

in ENV variables? since it's complaining about no API key

Brain isn’t quite working today. I have a flow Parameter called

cluster

which determines whether a cluster is used which is accessed from a callback to

DaskCluster

. However, it’s not actually used in the flow which is causing flow run failures locally because it’s an ‘unexpected parameter’. I am 90% sure I’ve seen a way before where I can say to Prefect to relax and that it’s fine (technical terminology), but can’t remember exactly how

Hey, I am getting this error when trying to register a flow. I am using protobuf for parsing input and output.

TypeError: cannot pickle 'google.protobuf.pyext._message.MessageDescriptor' object

Hello everyone, I am using a dask cluster and I've run into this issue:

distributed.protocol.pickle - INFO - Failed to serialize <Success: "Task run succeeded.">. Exception: cannot pickle 'lxml.etree.XMLSchema' object

which leads to

distributed.worker - ERROR - failed during get data with <ip> -> <ip>

which at some point close the connexion

distributed.comm.core.CommClosedError: in <TCP (closed)  local=tcp://<ip> remote=tcp://<ip>>: Stream is closed

this is managed by prefect by some retries (depending on configuration) and finally the agent raises an error and the flow is marked as failling

distributed.scheduler.KilledWorker: ('concatenate_df_based_on_time-b91c06dc30f54c5084e9f5fe8b6b32a5', <WorkerState 'tcp://<ip>', status: closed, memory: 0, processing: 1>)

Do you have an idea on how to prevent this kind of error?

Is the prefect2 setup for AWS ECS kind of the same as it was for prefect1? I followed this tutorial once and it was super-smooth for prefect1 https://towardsdatascience.com/serverless-data-pipelines-made-easy-with-prefect-and-aws-ecs-fargate-7e25bacb450c

Hey everyone! Can anyone tell me if the Orion API and/or Scheduler and/or UI support running multiple replicas? Is anyone running them in an HA config already?

Hi, when creating a class in a separate library. What is best practice so that the prefect logger picks up the log lines? Currently I'm on this path but I'm struggling to see them in the gui.

from logging import getLogger

logger = getLogger(__name__)

Class MyClass:
    def __init__(self):
         pass

    def my_method(self):
         <http://logger.info|logger.info>(‘hi’)

Does that make sense?

Hi. Is it possible to restrict number of parallel map tasks with DaskExecutor?

Our Team uses the task BigQueryLoadGoogleCloudStorage to load in blob uris to Bigquery. The docstring specifies that uri is an optional string https://github.com/PrefectHQ/prefect/blob/master/src/prefect/tasks/gcp/bigquery.py However, uri is used as the source_uris parameter for the load_table_from_uri method of the bigquery client. And according to the gcp bigquery client documentation, that parameter can be either a string or an iterable of strings https://cloud.google.com/python/docs/reference/bigquery/latest/google.cloud.bigquery.client.Client#google_cloud_bigquery_c[…]ent_load_table_from_uri Is there any reason why the prefect Task BigQueryLoadGoogleCloudStorage has it specified that its an optional string and not Union[str, Sequence[str]] ? Looking at the code it looks like it should be fine, but I want to make sure we dont fall into an anti-pattern.

Hi, I was wondering if there was a way to stop a scheduled flow if it failed X number of times?

What’d cause this error? I’m using an eagerly loaded template now (that’s in the project tree) as opposed to an S3 file path for the

job_template_path

now. Same template loaded via S3 during flow init didn’t throw this error, but registration blows up with this error now.

I've been looking at the nice

prefect_test_harness

in 2.0 and am wondering if any such fixtures exist for 1.2? Are there any general guides for setting up tests for projects using prefect?

Hey, just noticed that the API key i created last month doesen't show up in the prefect cloud API key tab. I can still see it whith

prefect auth list-keys

command but not in prefect cloud. Is it normal ?

Can you set task dependencies across flows? 🧐 We have some models that depend on prod models in our main flow but it would be nice to separate them into their own pipeline- however I don't want an unrelated task failure to prevent the dependent pipeline from running, so it would be better to set it on a task level

I am looking to setup Docker storage with multiple flows. I found this example in docs: https://docs.prefect.io/orchestration/recipes/multi_flow_storage.html but this example assumes all flows are in the same file, and there is no way to specify the file_path in

storage.add_flow()

method. Is there a way around it?

We are using prefect cloud for our org. If we use Git or Github storage, I believe the GITHUB_ACCESS_TOKEN is going to be stored on prefect cloud server? That’s basically like allowing prefect access to our private repos. Is there a way to avoid storing the access token on the cloud?

What’d cause a KubernetesRun to get stuck in this state?

Hi all - I'm trying out Orion w/ Kubernetes + Dask, and Google Cloud Storage for storage. I'm running into a pickling error (see thread) - any idea what I'm doing wrong?

Hey guys! Awkward new-guy first question incoming: I just installed Orion (

pip install prefect>=2.0b

) into an environment with mypy, and since there doesn’t appear to be a

py.typed

file in

prefect

, mypy is skipping it:

Skipping analyzing "prefect": module is installed, but missing library stubs or py.typed marker

I threw a py.typed in there (

touch py.typed

), but obviously this is less than ideal. Am I missing something obvious?

One more question! How can I specify resource requirements for Kubernetes in Orion? In v1 I used the JOB_MEM_REQUEST env variables (and similar): https://github.com/PrefectHQ/prefect/blob/master/src/prefect/agent/kubernetes/deployment.yaml#L36, but those don't seem to have an effect on the Orion agent.

Hi. Im creating my first flow - which is an daily ETL flow that reads data from csv-files and writes the data to big query. But now i wonder if there are any recommended ways to safeguard that duplicates aren’t written to big query if the flow it’s executed twice. I was thinking of using the cached_key_fn to not rerun the write task but feel unsure if that’s how it’s supposed to do. (I would rather have the task to be skipped..)

Hey ! Thanks to Kevin, I'm trying out a method to use

loguru

logger in Prefect tasks as described here. At first glance it works pretty well but it looks like

DEBUG

logs aren't sinked into Prefect. Any idea what I could do ?

Hi, wrt keeping the database size and performance good, is there a way to set some sort of retention rate policy on the database that maintains the flows, flowruns, etc?

Hi folks, our prefect kubernetes agent sometimes fails to deploy flow runs due to a connection error - namely a ReadTimeoutError (more details in the thread) Note that the same agent is able to deploy tens to hundreds of flows every day but seems to hiccup with these errors, leading to flow runs never getting submitted on our kubernetes cluster.

I'm trying to start a prefect orion flow from a flask route handler but I have an error when calling the

flow

decorated function "MissingContextError: No settings context found" What am I missing ?

hey, i'm trying to start a flow from cli with param flag. I have a boolean param that i want to set to true so i use

--param bool_param=True

but

True

is treated as a String during flow execution. How can i pass a boolean param in cli for flow run ?

Hi!, I saw an update in Prefect 2.0 doc that R will also be supported as flow language, Will Golang also be supported?