What is the benefit of running mapped tasks besides the batched api request? I'm running into a scenario where I want to map over

run_deployment

--> get the results, and then map over another

run_deployment

with parameter values populated from return of the previous map task. It's turning out to be kinda inflexible, and I'm thinking of just not doing the mapping in favor of running a loop (iterables are only max 20)

Hi everyone, I am running prefect with docker with containers like agent, Orion etc. In the agent i have mentioned 2 queues. The problem is If i make a single inferencing request everything works file. But, if i trigger multiple inferencing request, queue goes unhealthy. Can anyone tell me what I am doing wrong here. in the logs i am getting this error, but i am not able to get the significance

prefect.agent - Server error '502 Bad Gateway' for url '<http://http>://192.168.243.24:4200/api/work_queues/ca949cf5-ae18-4705-8e6f-311688abdfb3/get_runs'

Can anyone please tell me how can i resolve this.

It seems like the Prefect 2 Cloud API is down right now. I’m getting errors using the

prefect

CLI, and I can’t login at https://app.prefect.cloud/

Saw an earlier thread about getting 401 error while trying to access /auth/login endpoint. But didn't see any resolution on that. Also, if I switch to requests.get, the response code will be 200 but not sure if i should use anything in the header as refresh_token and access_token. Feel something is left out of the document which is required to run the POST successfully. Any ideas?

def login():
    auth_token = 'pnu_xxxxxxxxxxxx'
    auth_header = {'Authorization': 'Bearer ' + auth_token}
    response = <http://requests.post|requests.post>('<https://api.prefect.cloud/auth/login>', headers=auth_header)

Hi everyone, I'm getting an error in all of my flows when running from Prefect Cloud. After ~9:50 mins (almost exactly) the execution of whatever task within the currently running subflow is running gets terminated. I see an error that's something like this:

Crash detected! Execution was cancelled by the runtime environment.

ERROR
Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.9/asyncio/base_events.py", line 634, in run_until_complete
    self.run_forever()
  File "/usr/local/lib/python3.9/asyncio/base_events.py", line 601, in run_forever
    self._run_once()
  File "/usr/local/lib/python3.9/asyncio/base_events.py", line 1869, in _run_once
    event_list = self._selector.select(timeout)
  File "/usr/local/lib/python3.9/selectors.py", line 469, in select
    fd_event_list = self._selector.poll(timeout, max_ev)
  File "/usr/local/lib/python3.9/site-packages/prefect/engine.py", line 1613, in cancel_flow_run
    raise TerminationSignal(signal=signal.SIGTERM)
prefect.exceptions.TerminationSignal

The the subflow keeps running indefinitely despite having a failed task, and the main flow fails (its seems similar to https://github.com/PrefectHQ/prefect/issues/8481). This only seems to happen if the flow runs for 9 mins and 50 seconds, and it happens to whichever subflow/task is running at that time. Is there anything that might be causing this? Any help is greatly appreciated.

Good morning folks! I had posted an inquiry on #prefect-getting-started with no response, hence thinking that may not have been the proper place for it.. I'm moving the question here, in hopes someone is able to shed some light on it 🤷 --- Heya! I've got a flow that needs some troubleshooting / testing. How would I go about running a flow without it registering a run with prefect cloud? I could of course create a new profile or run the flow in a docker container with no

cloud login

done before, but that will prevent me from retrieving blocks i.e. secrets from the cloud workspace I don't want that flow to get registered on 😞

python flows/flow.py

In other words, the

@flow

decorated function make use of blocks available on a specific cloud workspace, but I want the flow run not appear on said workspace. I do not want anything sent back to cloud (for various reasons)

Hi, I try to build a deployment and upload to minio storage(remote-file-system), the package s3fs already install and i put this lines to head of flow.py

from prefect.filesystems import RemoteFileSystem
remote_file_system_block = RemoteFileSystem.load("minio-s3")

I get the error:

root@fd9f23f7b06c:~/flows# prefect deployment build rest_flow_3.py:websites2 -n eli -q check_url_status -sb remote-file-system/minio-s3
Found flow 'test url check flow'
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 230, in coroutine_wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 181, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/usr/local/lib/python3.8/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/usr/local/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/usr/local/lib/python3.8/site-packages/prefect/cli/deployment.py", line 988, in build
    deployment = await Deployment.build_from_flow(
  File "/usr/local/lib/python3.8/site-packages/prefect/deployments.py", line 755, in build_from_flow
    await deployment.upload_to_storage(ignore_file=ignore_file)
  File "/usr/local/lib/python3.8/site-packages/prefect/deployments.py", line 600, in upload_to_storage
    file_count = await self.storage.put_directory(
  File "/usr/local/lib/python3.8/site-packages/prefect/filesystems.py", line 368, in put_directory
    self.filesystem.put_file(f, fpath, overwrite=True)
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 114, in wrapper
    return sync(self.loop, func, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 99, in sync
    raise return_result
  File "/usr/local/lib/python3.8/site-packages/fsspec/asyn.py", line 54, in _runner
    result[0] = await coro
  File "/usr/local/lib/python3.8/site-packages/s3fs/core.py", line 1101, in _put_file
    await self._call_s3(
  File "/usr/local/lib/python3.8/site-packages/s3fs/core.py", line 332, in _call_s3
    await self.set_session()
  File "/usr/local/lib/python3.8/site-packages/s3fs/core.py", line 492, in set_session
    self.session = aiobotocore.session.AioSession(**self.kwargs)
TypeError: __init__() got an unexpected keyword argument 'url'
An exception occurred.

my setting is:

{
  "url": "<https://172.23.0.5:900>",
  "accessKey": "!!!!!!!!!!!!!!!!!",
  "secretKey": "===============",
  "api": "s3v4",
  "path": "auto"
}

Hi Community, I'm not sure if I'm stating something obvious here or whether this is thourougly documented. I have tried to search on this for a long time but to no avail. We are running our machine learning models as data pipelines with data ingestion > training > prediction. However, instead of creating a new flow/deployment back test for each of these models, I would prefer to create only one flow where each of the models can be imported using paramatization. How would you approach this task?

Hi 👋 Any ideas how to make a task fail from with in a callback? Calling raise Exception inside the callback won’t cause the task to fail.

Hi, I see a lot of emails with

Late run count:

messages, when I click on the link in the email it takes me to the work queue, but it is not so clear what task was late, how late and why? any thoughts around this ? the email looks like:

Name: <my cloud machine tag>
Last polled: 2023-02-12T21:34:05.094633+00:00
Late run count:
URL: <Work Queues tab>

Hi everyone, i want to create flow run task in the loop and the task is assigned paramater that is the day in between start and end date!

Hello everyone, I am getting the following error for one of the flows on the Prefect Cloud

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': 'Operation timed out', 'extensions': {'code': 'API_ERROR'}}])
Traceback (most recent call last):
  File "/opt/venv/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 91, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/opt/venv/lib/python3.8/site-packages/prefect/client/client.py", line 1917, in set_task_run_state
    result = self.graphql(
  File "/opt/venv/lib/python3.8/site-packages/prefect/client/client.py", line 569, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': 'Operation timed out', 'extensions': {'code': 'API_ERROR'}}]

The other flows are running without any errors. Has someone faced this issue before and find a resolution? Any and all help will be appreciated 🙇 Prefect Version : 0.15 CC: @Salim Doost @Jacob Blanco

Hi Prefect Community, question about Scaling Task.map method: It doesn't seem like that when using the map method for task it will scale to multiple K8s pods. Is there any way to make it do so? The only work around now is to just use sub flows inside for loops instead of using task map method if we want horizontal task scaling.

Hi Community, could you please check why some tasks keeps running forever? https://cloud.prefect.io/quiltai/task-run/c414a9a6-2f6c-4d3b-8f8b-a9d6eb835b45?logs

Are there any resources out there that explain how the prefect DAG-less approach works and the ideas behind it?

any1 else getting hit by

rash detected! Execution was interrupted by an unexpected exception: PrefectHTTPStatusError: Server error '500 Internal Server Error' for url '<https://api.prefect.cloud/api/accounts/36a3755f-b0f8-4d31-a7f7-745ce0d09e2e/workspaces/7a6aca52-8897-4fec-a53a-0102476b0ac1/task_runs/>'
Response: {'exception_message': 'Internal Server Error'}
For more information check: <https://httpstatuses.com/500>

been getting this alot

Hi Community, I am trying to use Prefect REST API to create flow run for a deployment but it's giving 403 and an error message saying that "Not authenticated" in response. What should I add for authentication ? Could any one help me on this? Thanks in Advance.

Hi all, I am running a deployment that runs a flow on defined queue every hour which fetches the data for the last hour [say

start_time

optional param set to

currenttime - 1 hour

by default]. Is it possible to run a custom run with start_time set to say 1 year prior to current time under a new queue ? The idea is to spin up new agents with higher specs to process more data for the entire year.

Hi all - is there a limit set on tags? Is it doable to set tags to group each user-related flows? Ty!!

Hi Community. At the moment I try to configure a local prefect orion server whoch shoud work with https : 443. I set the PREFECT_ORION_API_HOST, PREFECT_ORION_API_PORT=4200 and the PREFECT_API_URL=https://xxxxxxxx:4200/api but if i access the ui I get the message "Can't connect to Orion API at https://xxxxxxxx:4200/api Any hints what i forgot?

so prefect should run behind apache reverse proxy

RSVP for our Live Demo with Noteable 💻 When: Thursday @ 1 pm EST Where: RSVP here We will demonstrate how you can simplify the process of migrating ETL code to production using Noteable notebooks. Adopting these philosophies into your development pattern can save your data and engineering teams an enormous amount of time, increasing your overall productivity.

Terminal state handler Is there any way (or a plan to enable) handling the terminal state of a flow run? i.e. run some custom code on flow run termination and do something with the terminal state. (Prefect v2)

Hi everyone. Prefect v1 question (switching as soon as we can!) I have flows that are being registered with a schedule via a deployment process. When I add a new schedule through the UI, the one registered via code is disabled. I can't find a reference to this behavior in documentation, does anyone have any information on it? Is there a way to add new schedules in this case while keeping the original one active? Thank you!

Hi Team/ Everyone Yesterday, while watching one of the DE zoomcamp videos, I was introduced to a tool called Orchestration. I was looking for some courses on udemy and linkedin Learning, but I couldn't find any that had any real-life examples of its utilisation. Would it be possible for those of you who are knowledgeable and experienced with this platform to point me in the right direction? Thanks! It would be extremely helpful if you could recommend some guides or books.

any help here is much appreciated

hi I raised the version of prefect. 2.6.8 -> 2.7.11 Then, I cannot call subflow from flow and it freezes before subflow is called. Has the specification changed? thank you

When i try to set schedule with rrule from CLI like

prefect deployment set-schedule name --rrule "FREQ=DAILY;INTERVAL=10;COUNT=5"

I get following Error

Traceback (most recent call last):
  File "C:\Users\x\AppData\Roaming\Python\Python311\site-packages\prefect\cli\_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "C:\Users\x\AppData\Roaming\Python\Python311\site-packages\prefect\utilities\asyncutils.py", line 230, in coroutine_wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\x\AppData\Roaming\Python\Python311\site-packages\prefect\utilities\asyncutils.py", line 181, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\site-packages\anyio\_core\_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\site-packages\anyio\_backends\_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\asyncio\runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\asyncio\base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\site-packages\anyio\_backends\_asyncio.py", line 287, in wrapper
    return await func(*args)
           ^^^^^^^^^^^^^^^^^
  File "C:\Users\x\AppData\Roaming\Python\Python311\site-packages\prefect\cli\deployment.py", line 277, in set_schedule
    rrule_schedule = json.loads(rrule_string)
                     ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\json\__init__.py", line 346, in loads
    return _default_decoder.decode(s)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\json\decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python311\Lib\json\decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
An exception occurred.

I can make it work with interval and cron - what am i doing wrong here?

Hello Community. We have a strange issue - the kubernetes agent (2.8.0, image_: prefecthq/prefect:2.8-python3.8_) is connected with Cloud API (work queue is “Healthy”), it even successfully gets the pending flow run but during the execution it can’t update the state of the flow run with “403 Forbidden” error. Attaching the log from the agent. Any ideas, hints how debug/resolve this?

I’ve been using prefect in production for weeks (I’m not using prefect cloud) and today my prefect server seemed very slow. Looking at the logs I saw these warnings:

12:22:49.567 | WARNING | prefect.orion.services.failexpiredpauses - FailExpiredPauses took 5.385518 seconds to run, which is longer than its loop interval of 5.0 seconds.
12:22:49.881 | WARNING | prefect.orion.services.flowrunnotifications - FlowRunNotifications took 5.689214 seconds to run, which is longer than its loop interval of 4 seconds.
12:22:50.319 | WARNING | prefect.orion.services.marklateruns - MarkLateRuns took 6.143901 seconds to run, which is longer than its loop interval of 5.0 seconds.
12:22:55.451 | WARNING | prefect.orion.services.marklateruns - MarkLateRuns took 5.129949 seconds to run, which is longer than its loop interval of 5.0 seconds.
12:22:55.669 | WARNING | prefect.orion.services.failexpiredpauses - FailExpiredPauses took 6.088298 seconds to run, which is longer than its loop interval of 5.0 seconds.

Looks like there are some timeouts happening. I’m using a DNS for the prefect server, and both the prefect server and agents are pointing to this DNS. So the requests might be slow due to it. My question would be, how can I avoid these warnings (increase the timeouts)? I’ve searched the prefect settings (https://docs.prefect.io/api-ref/prefect/settings/) for envs which have default values of

4

and

5

seconds (as seen in the logs) and I found this ones:

PREFECT_ORION_API_KEEPALIVE_TIMEOUT
PREFECT_ORION_SERVICES_PAUSE_EXPIRATIONS_LOOP_SECONDS
PREFECT_ORION_SERVICES_LATE_RUNS_LOOP_SECONDS
PREFECT_ORION_DATABASE_CONNECTION_TIMEOUT

Increasing them to 20 seconds did not seem to make the warning dissapear. Any other suggestions? Behavior observer with prefect docker images 2.7.8 and 2.8.0. Thanks for your awesome support in advance 🙂