Anyone had an issue like this when a Flow is restarted by Lazarus? (it was a Spot instance that got yanked away)

ValueError: dictionary update sequence element #0 has length 1; 2 is required

Hi guys, does anybody knows why the Flow Run Logs in prefect UI doesn't reflect anything after "Beginning execution...." I can see additional log lines in my terminal - but it doesn't show in prefect 🤔

Hey everyone, I keep getting very weird errors when trying to scale my workloads in a Kubernetes cluster. Large amounts of flow runs fail with the following error:

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1103, in begin_task_run
    return await orchestrate_task_run(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1186, in orchestrate_task_run
    state = await propose_state(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1484, in propose_state
    raise prefect.exceptions.Abort(response.details.reason)
prefect.exceptions.Abort: This run has already terminated.

This really tells me absolutely nothing, anyone have an idea maybe? A second error that I see again and again is this one:

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/anyio/_core/_sockets.py", line 186, in connect_tcp
    addr_obj = ip_address(remote_host)
  File "/usr/local/lib/python3.10/ipaddress.py", line 54, in ip_address
    raise ValueError(f'{address!r} does not appear to be an IPv4 or IPv6 address')
ValueError: 'api.prefect.cloud' does not appear to be an IPv4 or IPv6 address

Which again seems super cryptic to me 😅 Any hints to where I could start digging would be greatly appreciated, thanks 🙂

Hi all 🙂 I am experimenting with a setup running the prefect-server in kubernetes, and having another container run an agent. But before running this agent, using

prefect agent start -q 'foo'

for example, I’d like to create and apply deployments using the

Deployment.build_from_flow

syntax described here https://docs.prefect.io/concepts/deployments/#create-a-deployment-from-a-python-object . Then I want this python script to run before I start the agent…

from prefect.deployments import Deployment
from flows.deployments.example_script import flow_name


deployment = Deployment.build_from_flow(
    flow=flow_name,
    name="Test",
    work_queue_name="test",
)

deployment.apply()

I’ve tried running it standalone and in Docker but I get a Attribute Error

module 'flows.deployments.example_script.flow_name' has no attribute 'name'

If I use

setattr(flow_name, 'name', 'something)'

before the definition of

deployment

then I don’t get this exact error but this

ValueError: Could not determine flow's file location.

I’m unsure of where to begin and would love any sort of help

Hello. My question is not directly related to Prefect 2, but it worked fine with Prefect V1. I am getting the following RuntimeError on import because one of my modules tries to call

asyncio.get_event_loop()

.

RuntimeError: There is no current event loop in thread 'AnyIO worker thread'.

Do you have any idea how to avoid it?

Hello all. I recently updated from Prefect 2.1.0 to 2.3.2. After updating, I re-deployed my existing deployments. I am now seeing flows being scheduluded twice. I do not see duplicate deployments listed in the UI or API. Based on the created dates, it appears Prefect is still scheduling using the 2.1.0 deployments as well as the 2.3.2 deployments. The 'old' flow runs have null value the deployment_id field. FYI I am using Prefect cloud. Has anyone seen anything like this and/or have a recommended way of purging what seems to be previous Prefect deployments still being scheduled?

Hi all, I have a question regarding using great expectations with prefect in production environment, I have an agent running on EC2 and flow code stored in S3. However, whenever I deploy and run a flow that has great_expectation's

run_checkpoint_validation

I get the following error but it runs fine locally. Also, the s3 storage used to store flow code has the great_expectations directory in it..

Hi, I’m having trouble getting prefect annotations to work with asyncio. I get the code to run fine but mypy complains that the annotated functions are no longer asynchronous. Removing either prefect annotations or typings allows it work Example:

import asyncio
from prefect import flow, task


@task
async def slow_task(a_value: int) -> int:
    await asyncio.sleep(1)
    return a_value


@flow
async def my_flow(a_value: int) -> int:
    return await slow_task(a_value)

if __name__ == "__main__":
    result: int = asyncio.run(my_flow(42))
    print(result)

results from mypy:

test.py:13: error: Incompatible types in "await" (actual type "None", expected type "Awaitable[Any]")
test.py:17: error: Argument 1 to "run" has incompatible type "None"; expected "Coroutine[Any, Any, <nothing>]"

Any tips on making typings work with asyncio and prefect

Hi. Why Prefect 2 runs flow in a separate thread? I have an issue with

tornado

that tries to get current event loop and fails.

Hi all, I was looking through the documentation of Prefect but could not easily find the answer to my two questions, so I hope you all can help me. Does Prefect support human interaction in the pipelines. e.g. data is being processed in the pipeline but must be validated manually before the final few steps. And does Prefect support states in the pipelines. Meaning the current run can be changed according to the last 10 runs? e.g. data is being processed and if the current value is greater than X and in the last 10 values there were at least 5 others with a value greater than X then I want the workflow to branch differently. I hope I explained it well enough

Hello all, I've noticed the following error intermittently using Prefect Cloud for the last few weeks. I'm running my agent on an EC2 instance. My deployments use docker containers. I updated recently from 2.1.0 to 2.3.2. The same errors have occurred with both versions. Any insight into why this may be happening?

prefect.exceptions.PrefectHTTPStatusError: Server error '500 Internal Server Error' for url '<https://api.prefect.cloud/api/accounts/043b2649-9d07-4c5e-8225-521ba2275e68/workspaces/689b139b-a725-4c2b-b167-86a705b8789d/task_runs/7d0c4bc0-a212-49b3-94e2-f1b9bdee765f/set_state>'
Response: {'exception_message': 'Internal Server Error'}
For more information check: <https://httpstatuses.com/500>

Hello All, When re-installing my helm chart to AKS via the Helm Chart, I’m running into this issue:

Error: failed to fetch <https://prefecthq.github.io/prefect-helm/charts/orion/prefect-orion-0.5.1.tgz> : 404 Not Found

The target version is: 2.3-python3.9. Has something changed to where this file no longer exists? That assumption doesn’t seem quite right.

Hi All, I am trying to implement a date parameter in prefect UI like so

start_date = Parameter("start_date", default="2022-09-11")

What I see on the UI after registering the flow is in the first attached image. start_date shows up as string. What I’d like to see by default is in the second image, I’d like start_date to show up as date type instead of as string with the date time picker functionality. How can I achieve this?

Hey all, does anyone know how can I retrieve an Infrastructure object from Orion? I haven’t seen it covered in the docs

Has anyone figured out how to overwrite parameters in prefect 2 for a flow on a per schedule basis, the way you could in prefect 1? A lot of overhead in Prefect 2 right now if you just want to send a simple variable change for a flow depending on the time the flow runs. Currently, the only solution I've found is to have 2 separate deployments registered with different schedules and variables.

Do flows and tasks need to be in the same Python script in Prefect 2.0?

Hello, I have been testing out prefect 2.0, we have been using 1.0 for over a year and love it. I have a GKE autopilot cluster set up on GCP with an agent running in a workload. I have been testing out running a very simple flow, and everything is working well. However, when I look at the logs from the agent workload, it picks up the flow run, says the pod is starting, and then says the pod never started. The pods do start, the flow runs, and I see its successful in the UI. I am just wondering why the agent says the pod was never started. This is my first time using kubernetes btw so bare with me.

We have some flows that could run locally for small datasets. What is the recommended way to invoke a flow and dynamically change the associated TaskRunner?

This message was deleted.

Hey folks! I’m seeing this rather cryptic pickle error when running Prefect locally (Prefect 2.4.0, Python 3.10) - my code doesn’t use Pydantic’s

AnyUrl

or subclasses within the code, do you have any idea what might be causing the pickle error?

Hi, any suggestions on how to pass a dataframe as a parameter to a flow? This breaks in latest Prefect:

from prefect import flow
import pandas as pd
import numpy as np


@flow
def do_nothing(df):
    return None


if __name__ == "__main__":
    df = pd.DataFrame(np.random.random((10_000, 20)))
    do_nothing(df)

Traceback in thread

Hello! I have a question regarding how resources are being used in Prefect. I am using the DaskExecutor with LocalCluster(2 workers, and 4 threads per worker) and am testing processing 800 files which I am splitting evenly into 8 batches. Originally I had the prefect task running each of the batches passed to them and processing those in a loop, but when I change this to explicitly run threading within the tasks (still mapping to them) using the existing client like how Kevin suggested in this POST, I am seeing significant time savings (25 - 50% faster). My question is are there drawbacks to performing batch processing this way/is this the right way to do this or were we not utilizing Prefect in the correct way previously?

For those interested, I'm creating a Prefect Collection to interact with Jinja. I'm still working on it but I've already posted the basics on GitHub: https://github.com/hallenmaia/prefect-jinja

Does anyone know if it’s possible for one’s own Prefect Cloud account to kick off a flow that’s in another Prefect Cloud account? My org delivers data to another org that also uses Prefect and it’d be cool if once my data delivery finishes, it could automatically kick off their pipeline that handles said data. Maybe if not cross-cloud, there could be an email trigger (if the other org gave me their account and flow IDs)?

Hello Team, I am getting intermittent crashes on prefect without any reason being shown Is there any bug in the system causing this?

Hey all, is there a sync OrionClient? I’m developing a CLI with custom features and the sync client would be very helpful

Been trying to get the prefect helm chart working and getting a little confused

In Prefect 1.0, is it possible to schedule a flow, but not allow (disable certain roles) users to run it?

Hi. At last I'm properly getting to 2.0! 😒uccess-kid: Seems I'm running into the same issue as Keith here Getting

ModuleNotFoundError: No module named 'gcsfs'

It's strange because the environment that my agent is running in has

gsfs

installed and I have

"EXTRA_PIP_PACKAGES": "gcsfs"

in my Docker Block... Any tips?