I'm trying to install the vertex agent for Prefect with

pip install prefect[gcp]

as per the documentation but I'm getting

WARNING: prefect 2.6.4 does not provide the extra 'gcp'

. Maybe the docs are not up to date? Anyone know how to install the vertex agent now?

Curious if anyone has used prefect with real-time data pipeline curious what's your setup like? Also, has anyone used prefect to run Spark jobs?

My Prefect 2.0 cloud workspace shows up in my list, but is inaccessible with a 404 error. What do I do?

Hi guys, I have been using prefect 2.0 for a while for my multiple project and I have come across this error often

raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '403 Forbidden' for url '<https://api.prefect.cloud/api/accounts/xxxxxx>
    return asynclib.run(func, *args, **backend_options)

Usually re - running the agent works fine but given the sensitive nature of my prjects, I would like to avoid this in future, hence i am trying to understand why is this happening

Hi, I'm running a prefect 2.5.0 agent on a VM in azure. Occasionally when not running any jobs, like in the middle of the night, the agent throws the following exception when trying to ping the prefect api:

PrefectHTTPStatusError: Client error '403 Forbidden' for url
'<https://api.prefect.cloud/api/accounts/><ACCOUNTGUID>/wor
kspaces/<WORKSPACEGUID>/work_queues/<WORKQUEUEGUID>/get_runs'

At /usr/local/lib/python3.8/dist-packages/prefect/client/base.py:125. Submitting any flows after this exception leads to them being visible in the Prefect cloud web interface as "Scheduled" or "Late", basically the agent completely dies after this. Is this a known issue or could this be due to something else I'm doing incorrectly?

Hi, i'm running prefect 1.2.2. It was all running smoothly until i upgraded the kubernetes version (eks) from 1.21 to the latest 1.23 Now i'm randomly get the following error

INFO - Retiring workers [154, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172, 173, 174, 175, 176, 177, 178, 179, 180, 181, 182, 183, 184, 185]
INFO - Adaptive stop
INFO - Adaptive stop
ERROR - prefect.CloudFlowRunner | Unexpected error: KilledWorker('<name>', <WorkerState 'tcp://<ip>', name: 47, status: closed, memory: 0, processing: <number>, 3)

Restarting the flow is resolving the issue. Is there any sensible explanation of why upgrading kubernetes cluster could cause it? Or i must be missing something elsewhere ?

Hi, I'm running prefect 2.6.3. I have an agent in docker container running as ECS service and Github filesystem with private repo to store my flows. The flows are in 'flows' subfolder in repo. I'm getting this traceback when trying to run the flow:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 247, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.10/site-packages/prefect/client/utilities.py", line 47, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/deployments.py", line 159, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=deployment.path, local_path=".")
  File "/usr/local/lib/python3.10/site-packages/prefect/filesystems.py", line 146, in get_directory
    shutil.copytree(from_path, local_path, dirs_exist_ok=True)
  File "/usr/local/lib/python3.10/shutil.py", line 556, in copytree
    with os.scandir(src) as itr:
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmp2ywchethprefect/flows/hello_world'

Have you got experience with such issues?

V1 question - I'm registering a flow that has a schedule with multiple clocks. Even if I make no changes to the flow structure/metadata, registering always increases the version number. It is something to do with the schedule + clocks, but I have set the start and end dates to fixed values and the default parameters and cron values are not changing. Am I doing something wrong? I just want to be able to register the flow and not have it create a new version if nothing has changed

I'm running flows with docker infrastucture, but need to access files from a shared network drive inside the docker. Since i can't edit docker run command as it start with the flow run - is there any way i can mount the network drive in a way that my flows can access them inside the docker container?

I will start new thread since my last one was marked with a tick even though it's not resolved. I will rephrase it. In the docker block, i need to use the volume argument, currently it says

A list of volume mount strings in the format of "local_path:container_path".

Would following bind my Local G:/ Drive to a G:/ in the docker, or should I write it differently?

🤔 Using the default setup for result caching with Prefect 2.6.4 and I always get this exception at the end of a Completed Flow run...

Traceback (most recent call last):
  File "/home/tenders/Documents/code/prefect-orion/platform_prefect/accounts/accounts_subscriptions.py", line 359, in <module>
    flow_result = main()
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/prefect/flows.py", line 439, in __call__
    return enter_flow_run_engine_from_flow_call(
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/prefect/engine.py", line 150, in enter_flow_run_engine_from_flow_call
    return anyio.run(begin_run)
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/prefect/client/utilities.py", line 47, in with_injected_client
    return await fn(*args, **kwargs)
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/prefect/engine.py", line 229, in create_then_begin_flow_run
    return await state.result(fetch=True)
  File "/home/tenders/.cache/pypoetry/virtualenvs/prefect-orion-HonJDUqB-py3.10/lib/python3.10/site-packages/prefect/states.py", line 86, in _get_state_result
    raise MissingResult(
prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

It seems like Prefect is still expecting the State data to exist, even though it doesn't by default? (I may be misunderstanding things too)

Hey folks! I’d like to use python’s dataclasses and yamldataclassconfig packages to read job configurations from a yaml file. The code I have works fine outside of the context of a task, but when I wrap the code in a task I run into what looks like a fairly obscure error (in the thread). The implementation of the code is straightforward. I have a dataclass (partial version for illustration):

@dataclass
class Config(YamlDataClassConfig):
    EMAIL_FROM: str = None
    EMAIL_TO: List[str] = None

as well as a task intended to instantiate the instance of the class and load data to it:

@task(name='load-configs')
def load_configs(config_path: str) -> Config:
    configs = Config()
    configs.load(config_path)
    return configs

The task seems to instantiate the config object, but breaks when we try to actually load configs from the provided path. Again, this works perfectly fine outside of the context of a task, so presumably there is some internal prefect logic I’m futzing with here. I’d appreciate any thoughts!

Apologies if I’ve missed this somewhere obvious (quite likely!) but is it not possible to cancel a flow run from the 2.0 UI?

If you have a flow (C) that calls two different flows (A & B). Will prefect use the deployments for the subflows or the parent? I ask because w/ ECS the deployment is what references the ECS Block that contains the cpu & memory settings. There are times where we would want each subflow to run w/ different resources

Hi! We're starting to look into a transition to Prefect 2 so I have a few likely basic questions, I'm trying to test out getting a very simple from running with s3 storage and ECS infrastructure. I've gotten to the point where the task successfully launches, but I'm running into a couple issues I'll detail in the comments.

Hi all! I'm wondering if Prefect supports dynamic task definition & dependency assignment, like in the linked AirFlow code? https://github.com/konosp/dbt-on-airflow/blob/8de43f2fef38b8e3f34c86aba77ed3210379c527/airflow/dags/schedule_dbt.py#L73

hi team, im running a prefect flow which sends http requests to my server. when the requests last for 10-20 minutes, prefect cannot receive the response, I can see from my server log that a 200 is already returned, anyone has any idea?

prefect always fork a new thread to execute task, is there a way to run task in process instead of thread? As there are some library that handle

signal

by themselves, and due to this limitation

When threads are enabled, this function can only be called from the main thread of the main interpreter; attempting to call it from other threads will cause a

ValueError

exception to be raised.

https://docs.python.org/3/library/signal.html#signal.signal It will fail when execute as task of prefect.

I am observing this behaviour with my prefect agent I am running a agent from EC2 cloud instance ( Ubuntu ) After I leave the screen , the agent run few successfully flow runs ,but after that the agent just seems to pauses. From the prefect ui , i can see the flows goes into a 'late' state , and when i again connect to my EC2 instance and open the terminal screen running my prefect agent, the agent starts running everything imedieatly causing a hike in memory computation and unstacking all the late flows. I have observed this bheaviour of agent with multiple projects running on EC2

Hi all. Is there a way to filter out default prefect logs like:

Starting task run...
[2022-10-25 15:23:07+0100] INFO - prefect.TaskRunner | Task 'read_features_config[1]': Starting task run...
[2022-10-25 15:23:07+0100] INFO - prefect.TaskRunner | Task 'read_features_config[3]': Finished task run for task with final state: 'Success'

hello, why my flow always stuck at Pending State with the agent log like:

I found the explaination here, but seems like it's for prefect 1, there is no --show-flow-logs in prefect 2

hi! im trying to use my postgres db running on rds as the database for orion(self hosted instance) backend. but it continuously runs into time out errors. has anyone implemented this before or facing the same issue ?

Hello all, does anyone know what the Prefect Cloud retention policy is for flow and task results, and how large the

location

field may be (i.e. for use with

PrefectResults

)? Thanks!

Hi team, I’m a bit lost in how to configure JSON logging in Prefect 2.0. I see this concept document and this api reference on the JSONFormatter, but I can’t seem to find how to marry the two together. What we’re trying to achieve is: 1. Prefect Cloud 2.0 still shows logs as is 2. Kubernetes jobs output logs in JSON so that they can be ingested by datadog (we want each log entry to have the flow_name, task_name, etc. to make it easily searchable back in time).

Hi folks, quick dumb question : to set to the type of parameter for a flow, do i just use type hints to set the type of parameters?

Hi All, I keep getting the below Timeout error while fetching data using graphql in a python code for Prefect 1.0 prefect.exceptions.ClientError: [{'path': ['flow_run'], 'message': 'Operation timed out', 'extensions': {'code': 'API_ERROR'}}] Is there any solution to avoid this?

Hello, I'm trying to deploy prefect agents on an on prem environment, but I get a

prefect.exceptions.PrefectHTTPStatusError: Client error '403 Forbidden' for url '<https://api.prefect.cloud/api/accounts/xxxxxxxx/workspaces/xxxxx/work_queues/name/dev-prem-wq>'

I tried changing the api keys (service account) / giving ownership to the agent on the workspace, but nothing worked so far. I tried deploying the agents on the cloud on a kubernetes cluster and it works just fine, but when I try to do the same thing on our on prem environment, it returns me this error. Can I have some insights please ? Thanks

@Nate Hi Nate, I suppose to have a meeting with you tomorrow at 8PST I did not get the email invite from https://calendly.com/ (yet) Youval

Fugue is a unified interface for distributed computing that lets users execute Python, pandas, and SQL code on Spark, Dask and Ray without rewrites. prefect-fugue provides the Fugue tasks, context and blocks for Prefect. In Prefect Live at 3PM Eastern today, Kevin Kho, a former community engineer at Prefect, will talk about how to use this integration and its use cases. Come join us live on Twitch. Subscribe to Prefect Community Calendar, available at prefec.tv/calendar, to get updated with all Prefect Live events.