Hi there! We are working on ML applied to computer vision, therefore our pipelines take long time and deal with "_heavy_" data that's best kept close to the training code. So far we have been babysitting the processing and training pipelines, but that's of course not scalable and we are starting to take too long time. I am currently investigating how we can make our pipelines fully automated, reproducible and traceable. However, I am not finding many examples of Prefect applied to image processing and computer vision on the internet. I am concerned with the fact that images are batch-processed but most often we actually want to stream, process and load to another location. Using Prefect's

.map

seems like a good choice, but won't it make the

Flow Runs UI

cluttered and not usable ? In your experience, how do you see people solve this problem ?

Hi Prefect Team - On our company's dashboard for Prefect 1 it appears that all the flow schedules have been turned off. The flows are running still, but this caused us some confusion this morning.

Hello Everyone, this might seem like a trivial question, but I'm finding it hard to figure out if I need to be using Agents or Workers, most of our deployments will be running on a kubernetes infrastructure block. Thanks

Hello Everyone, I’m currently migrating our infra from Prefect 1 to Prefect 2. In Prefect 1 I would create a

Flow Config

and assign

Storage

to a

Docker

type and pass it the ECR registry_url, image_tag, etc. When trying to configure the same thing for Prefect 2 I came across the Docker Registry Block that I would assume it would be compatible with ECR but it asks me for username and password credentials while we use rotating creds as authorization tokens for ECR are usually temporary. What are the best practices in Prefect 2 to communicate and retrieve the flow code stored in Amazon ECR?

Hi all, A couple of questions: 1. Anyone could point me to what is the functionality in prefect 2.0 that does the same as

@task(LocalResult(dir='path/to/folder/for/storage),target='result.pkl',checkpoint=True)

2. In prefect 1.0, if a task returns a dictionary and I specify as above how to locally cache why may be getting that

result.pkl

only saves the last element in the dictionry? Thanks,

Hi all - in our K8s deployment I'm often seeing a long list of "RUNNING" flows where the underlying pod has crashed (often due to OOM). Any ideas why these runs aren't reflecting a "CRASHED" state in the API?

hi there. after adding a GitHub credentials block via UI, I updated my prefect.yaml pull section to

access_token: "{{ prefect.blocks.github-credentials.example-token }}"

but my agent is unable to retrieve this properly and I receive a

"fatal: unable to access.. URL using bad/illegal format or missing URL"

message. what am I doing wrong here? thanks in advance

Hey Team - is there any more information on the query syntax? I'd like a query which gives me all of the flows which currently have scheduled runs but do NOT have their schedule turned on

MS Teams web hook automation only posts Completed notifications for some workflow runs, even though each log shows it ending in state Completed and the automation is set notify upon entering "any state".

22:40:17.227 | INFO  | Flow run 'peach-flamingo' - Finished in state Completed('All states completed.')

22:40:19.792 | INFO  | prefect.infrastructure.process - Process 'peach-flamingo' exited cleanly

has anyone experience async flows hanging for no apparent reason? I’m doing a flow/sub-flow structure where the flows can spawn other flows (as sub-flows) and for some reason at some point in execution everything just stops - no more logs sent or anything. I feel like I’m missing something obvious with how async flows/tasks work within prefect but curious if anyone has encountered this before?

Hi All, My project contains multiple Prefect2 deployments. As a storage I use S3, but I've noticed that each time I build 1 deployment, all of them are being re-uploaded to S3. Is there any way to restrict the upload to the 1 I build right now?

Hello everyone, because of miscommunication we haven't upgraded our subscription in Prefect 1 to avoid freezing, while we work on migrating to Prefect 2. Anyone had a similar case or knows, how we can upgrade our account now to unfreeze it and buy us some time?

Hey guys! Huge fan! Prefect is very cool and fun to use

Lately I’ve been running into some issues and I was wondering if the experts here could help me out

so I’m using

prefect.infrastructure

’s

DockerContainer

and I want to set its

image_pull_policy

to

ALWAYS

but for some reason it fails to pull the image from the registry

I’ll paste the error in a second but yeah, would be great if someone has already experienced something like this

Hey everyone, Is there any way i can gradually increase the concurrency of aa flow in prefect.

Is there someone from prefect security team here?

Hello again guys, sorry to bother but I came across another issue and couldn’t find a way to solve it. I’m trying to spin a Prefect Agent in a Docker Container, setting up the correct env vars in the form of:

PREFECT_API_KEY=[my_key]
PREFECT_API_URL=<https://api.prefect.cloud/api/accounts/{account_id}/workspace/{workspace_id}>

When I try to spin up the container, I always get a not found

Response: {'detail': 'Not Found'}

response even though it exists.

prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<https://api.prefect.cloud/api/accounts/{account_id}/workspace/{workspace_id}/flow_runs/filter>'

If I spin the command (in this case

prefect agent start --pool template-dev --work-queue dev

) in my local machine outside a container it works fine. I have the same values I get in

prefect config view

while spinning up the Docker Container. Can anyone help with this?

Hi All, sorry to bug about an issue with an on prem setup again but we're updated an on prem prefect server to use postgres, updated the PREFECT_ORION_DATABASE_CONNECTION_URL (we're on 2,7,12) on a different profile to the correct url, but Prefect is using the sqlite database instead, even when we use the profile using a call to start the server with the '-p profile_name' flag. Are there any other flags to set? I can't see what the setting is on the server as it's blanked out, which makes sense as it could have passwords in it.

Prefect 1.0 self hosted I'm encountering an issue where flows that are registered with a schedule, with

set_schedule_active=False

, won't run on schedule (will always be "Late"); and, when clicking on the "Run now" button on the UI on this flow (in the window of scheduled runs), also won't run. Any ideas?

Hi everyone. I am using this recipe to run a worker on ECS. The worker starts and communicates with Prefect Cloud. However, it has crashed repeatedly (7 times in the last 12 hours or so). The error I see is

RuntimeError: Service exceeded error threshold.

Any ideas? I am attaching the logs in the thread.

Version:             2.10.9
API version:         0.8.4
Python version:      3.10.8
Git commit:          1655c1fa
Built:               Thu, May 11, 2023 2:29 PM
OS/Arch:             linux/x86_64
Profile:             cloud
Server type:         cloud

Prefect 2.10.7 self hosted Looks like my ECS task infra block is not honouring the entrypoint offered in the deployment. Instead the

python -m prefect.engine

command just installs the python libraries, and then stalls in

Pending

state. Any pointers here?

I've run into the "too-many-revisions" issue using ECS Task Block in Prefect2. I've removed everything from the Task Definition property and added an explicit Task Definition ARN, but I'm still getting new revisions of the definition. Is there something I'm missing to prevent new revisions?

Hi Community, We need help with Prefect agent setup. We have hosted Prefect server in EC2 instance and created prefect agent in ECS Faregate. We are using docker image as “prefecthq/prefect:2.10.8-python3.10”. We configured environment variables "‘PREFECT_API_URL’ , PREFECT_CLOUD_API’ ‘PREFECT_LOGGING_LEVEL’INFO’,‘AWS RETRY MODE’‘adaptive’,‘AWS_MAX_ATTEMPTS’:‘10’ in ECS Fargate Agent Service and run ""prefect agent start -p default-agent-pool -q default"" command through command line. In ECS Fargate while running services for agent we are getting below error. Error logs : Agent started! Looking for work from work pool ‘default-agent-pool’… Failed the last 3 attempts. Please check your environment and configuration. Examples of recent errors: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 111, in connect_tcp stream: anyio.abc.ByteStream = await anyio.connect_tcp( File “/usr/local/lib/python3.10/site-packages/anyio/_core/_sockets.py”, line 213, in connect_tcp async with create_task_group() as tg: File “/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py”, line 658, in aexit raise CancelledError asyncio.exceptions.CancelledError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 10, in map_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 110, in connect_tcp with anyio.fail_after(timeout): File “/usr/local/lib/python3.10/site-packages/anyio/_core/_tasks.py”, line 118, in exit raise TimeoutError TimeoutError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 60, in map_httpcore_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 353, in handle_async_request resp = await self._pool.handle_async_request(req) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 253, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 237, in handle_async_request response = await connection.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 86, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 63, in handle_async_request stream = await self._connect(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 111, in _connect stream = await self._network_backend.connect_tcp(**kwargs) File “/usr/local/lib/python3.10/site-packages/httpcore/backends/auto.py”, line 29, in connect_tcp return await self._backend.connect_tcp( File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 109, in connect_tcp with map_exceptions(exc_map): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 14, in map_exceptions raise to_exc(exc) httpcore.ConnectTimeout The above exception was the direct cause of the following exception: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/prefect/utilities/services.py”, line 46, in critical_service_loop await workload() File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 196, in get_and_submit_flow_runs work_queue_names=, File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 196, in work_queue_names=, File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 144, in get_work_queues work_queue = await self.client.read_work_queue_by_name( File “/usr/local/lib/python3.10/site-packages/prefect/client/orchestration.py”, line 844, in read_work_queue_by_name response = await self._client.get( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1754, in get return await self.request( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1530, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 246, in send response = await self._send_with_retry( File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 192, in _send_with_retry response = await request() File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1617, in send response = await self._send_handling_auth( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1645, in _send_handling_auth response = await self._send_handling_redirects( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1682, in _send_handling_redirects response = await self._send_single_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1719, in _send_single_request response = await transport.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 352, in handle_async_request with map_httpcore_exceptions(): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 77, in map_httpcore_exceptions raise mapped_exc(message) from exc httpx.ConnectTimeout Failed the last 3 attempts. Please check your environment and configuration. Examples of recent errors: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 111, in connect_tcp stream: anyio.abc.ByteStream = await anyio.connect_tcp( File “/usr/local/lib/python3.10/site-packages/anyio/_core/_sockets.py”, line 213, in connect_tcp async with create_task_group() as tg: File “/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py”, line 658, in aexit raise CancelledError asyncio.exceptions.CancelledError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 10, in map_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 110, in connect_tcp with anyio.fail_after(timeout): File “/usr/local/lib/python3.10/site-packages/anyio/_core/_tasks.py”, line 118, in exit raise TimeoutError TimeoutError During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 60, in map_httpcore_exceptions yield File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 353, in handle_async_request resp = await self._pool.handle_async_request(req) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 253, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py”, line 237, in handle_async_request response = await connection.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 86, in handle_async_request raise exc File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 63, in handle_async_request stream = await self._connect(request) File “/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py”, line 111, in _connect stream = await self._network_backend.connect_tcp(**kwargs) File “/usr/local/lib/python3.10/site-packages/httpcore/backends/auto.py”, line 29, in connect_tcp return await self._backend.connect_tcp( File “/usr/local/lib/python3.10/site-packages/httpcore/backends/asyncio.py”, line 109, in connect_tcp with map_exceptions(exc_map): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py”, line 14, in map_exceptions raise to_exc(exc) httpcore.ConnectTimeout The above exception was the direct cause of the following exception: Traceback (most recent call last): File “/usr/local/lib/python3.10/site-packages/prefect/utilities/services.py”, line 46, in critical_service_loop await workload() File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 261, in check_for_cancelled_flow_runs async for work_queue in self.get_work_queues(): File “/usr/local/lib/python3.10/site-packages/prefect/agent.py”, line 144, in get_work_queues work_queue = await self.client.read_work_queue_by_name( File “/usr/local/lib/python3.10/site-packages/prefect/client/orchestration.py”, line 844, in read_work_queue_by_name response = await self._client.get( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1754, in get return await self.request( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1530, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 246, in send response = await self._send_with_retry( File “/usr/local/lib/python3.10/site-packages/prefect/client/base.py”, line 192, in _send_with_retry response = await request() File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1617, in send response = await self._send_handling_auth( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1645, in _send_handling_auth response = await self._send_handling_redirects( File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1682, in _send_handling_redirects response = await self._send_single_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_client.py”, line 1719, in _send_single_request response = await transport.handle_async_request(request) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 352, in handle_async_request with map_httpcore_exceptions(): File “/usr/local/lib/python3.10/contextlib.py”, line 153, in exit self.gen.throw(typ, value, traceback) File “/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py”, line 77, in map_httpcore_exceptions raise mapped_exc(message) from exc httpx.ConnectTimeout Agent stopped! Thanks, Deep

Hi all, I have a question regarding serializers. I use the following in a task to save the result to a

.pkl

file. The result is a dictionary with two elements (two arrays).

@task(cache_key_fn=cache_file_based_fn,
      persist_result=True,         result_storage=LocalFileSystem(basepath=settings.main_dir+'/results'), 
      result_storage_key=settings.main_task_file,
      result_serializer="pickle",
      cache_result_in_memory=False)

Json is not supported because narrays cant be encoded with JSON. Pickle is fine with me but when I try to open the pickle file that was generated I get the following error:

UnpicklingError: invalid load key, '{'.

Any clue why this is happening? just FYI:

settings.main_task_file = 'results.pkl'

Posted in #prefect-getting-started

Hi again. I am trying to understand how to use projects to run flows on ECS. Could someone explain which recipe I should use, assuming I want to store code in GitHub? Also, could you direct me to some yaml examples? Thanks!

@prefect Team can someone provide me examples around how to implement flow code storage using github repo ,I did notice only githubrepourl/reponame ,how to include multiple paths

Hey team - on prefect 1, how are the parameters for each run passed into an ECS task definition? I do not quite understand. Any help on explaining the AWS side would be appreciated!

Hey team, just learned about Marvin and can't stop using it! That said, I'm now encountering a timeout error when I try to use an `@ai_model`to extract structured data from freetext, as below. Seems like it's set to 60 seconds somewhere:

13:23:45.510 | WARNING | langchain.chat_models.openai - Retrying langchain.chat_models.openai.acompletion_with_retry.<locals>._completion_with_retry in 1.0 seconds as it raised Timeout: Request timed out.