Hi all 🙂 Our current system uses FastAPI to handle requests, which trigger tasks in Celery. These tasks are placed in a queue in RabbitMQ, and Celery workers process them, performing ML computations and providing results. We use callbacks and monitoring to track the progress of these ML tasks. In addition, we have Spark Scala tasks that require a lot of computing power and run in a distributed cluster for data processing. Unlike Celery, Spark has its own methods for scheduling jobs and processing data. So, it’s important to manage resources carefully to avoid conflicts and performance issues between Spark Scala and Celery. One challenge we face is that Spark Scala and Celery tasks have different coordination and orchestration needs. Spark has its own built-in mechanisms, while Celery relies on RabbitMQ and worker pools. To address this issue, we want to try using Prefect. Prefect as a tool for managing workflows, which includes defining dependencies or scheduling. By using Prefect’s features, we hope to be able to orchestrate both Spark Scala and other tasks within a single workflow, simplifying the definition and management of our workflows. Do you have any feedback on such a goal? I saw this discussion. Also, we would like to know how difficult it would be to modify parts of our code to work with Prefect’s APIs and conventions vs our current task definitions. Saw this feedback from Jellyfish

@all I am facing this issue on Kubernetes helm deployment, Does anyone has a solution to it? Worker 'KubernetesWorker d7daa483-b7b2-412f-b8cd-c46ab538422d' started! Traceback (most recent call last): File "/usr/local/lib/python3.11/site-packages/prefect/cli/_utilities.py", line 41, in wrapper return fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 260, in coroutine_wrapper return call() ^^^^^^ File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 245, in call return self.result() ^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 173, in result return self.future.result(timeout=timeout) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 449, in result return self.__get_result() ^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result raise self._exception File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 218, in _run_async result = await coro ^^^^^^^^^^ File "/usr/local/lib/python3.11/site-packages/prefect/cli/worker.py", line 113, in start await worker.sync_with_backend() File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 624, in sync_with_backend await self._update_local_work_pool_info() File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 579, in _update_local_work_pool_info work_pool = await self._client.read_work_pool( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/site-packages/prefect/client/orchestration.py", line 2125, in read_work_pool response = await self._client.get(f"/work_pools/{work_pool_name}") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/site-packages/httpx/_client.py", line 1754, in get return await self.request( ^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/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.11/site-packages/prefect/client/base.py", line 280, in send response.raise_for_status() File "/usr/local/lib/python3.11/site-packages/prefect/client/base.py", line 137, in raise_for_status raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause prefect.exceptions.PrefectHTTPStatusError: Client error '405 Method Not Allowed' for url 'http://prefect-server:4200/api/work_pools/' Response: {'detail': 'Method Not Allowed'} For more information check: https://httpstatuses.com/405

hello all......I have prefect running in two servers........I want to achieve HA on this servers......i.e, Prefect should run in One Server and other should stay as standby and no business logic should run.....To achieve this I pointed the Metadata DB to Postgres DB........I see none of my servers running the prefect jobs........Can someone please help @all

Hi, I get this error : prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API. 051622 PM ERROR Finished in state Failed('Flow run encountered an exception. MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.\n') This is a frequent error but a temporary one in one of my flows.

Has anyone had their flow schedules interrupted recently on Prefect Cloud? I had some flows that stopped working on Friday. It looks like their schedule was disabled (but not by me), and yet when I looked in another part of the instrumentation it seemed like they were still scheduled to run as normal. Other flows still ran fine according to schedule, but just wondering why one in particular suddenly stopped and whether anyone else had experienced anything similar

Is this still the best way to run subflows in parallel? https://discourse.prefect.io/t/how-can-i-run-multiple-subflows-or-child-flows-in-parallel/96

hello community, hello prefect team, Prefect Agent rightsizing? We are running multiple prefect agents on aws ec2 instances and were wondering if there is any recommendation from prefect or experience in the community regarding cpu/memory requirements for the agent. how many flows should be managed by one agent at a time, what is the average resource consumption of a flow run on the agent and is there any known limit to the number of flow runs per agent? happy to hear from your experiences.

Hi everyone. I am trying to run a worker with a non-default work queue. I the

prefect worker start

command seems to be ignoring the queue. The worker starts and polls Prefect Cloud, but it seems to be connecting the default queue. I am starting the worker with

prefect worker start --pool local-pool --work-queue ingest-tracker

Has anyone else run into this?

Am having trouble loading a .env file using python-dotenv within Prefect. Within

Deployment.build_from_flow

, am setting

is_schedule_active=ENV == "production"

based upon whether the env is production, or not. This is determined by an import from another file:

from common.config import ENV

and in that file is:

# have also tried dotenv.find_dotenv() without success
dotenv_path = os.path.abspath(os.path.join(__file__, "../../../.env"))
if not os.path.isfile(dotenv_path):
    raise RuntimeError(dotenv_path)

This loads happily in a local ipython shell. But when run as a prefect flow, the folder structure appears to be different:

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 940, in exec_module
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "/private/var/folders/d8/sqvg_pms3pd0xvcl2tvvxhqr0000gn/T/tmp8e6a8h47prefect/prefect_daily_internal_metric.py", line 9, in <module>
    from common.config import ENV
  File "/private/var/folders/d8/sqvg_pms3pd0xvcl2tvvxhqr0000gn/T/tmp8e6a8h47prefect/common/config.py", line 9, in <module>
    raise RuntimeError(dotenv_path)
RuntimeError: /private/var/folders/d8/sqvg_pms3pd0xvcl2tvvxhqr0000gn/T/.env

Hi Team, Am reposting the issue again but am getting when migrating working prefect 1 flow to prefect 2.

psycopg2.ProgrammingError: no results to fetch

The flow has bunch of tasks which shares redshift connection across the tasks and it fails in the first task. Any idea why this could happen?

I have a flow that runs every 15 min. It's idempotent and if one run does not happen as scheduled, nothing bad happens. Also, there is not need to execute late flow runs. Is it possible to configure the flow/deployment so that if there is a bunch of late flow runs queued, it runs only one or skips all the late runs?

Hi all, I need help with a Python deployment. Whenever, I try to deploy the script to Prefect Cloud. I get the following error. ValueError: Unable to find block document named xxxx/xxxx for block type secret here is my code. deploy_flow = Deployment.build_from_flow( flow=process_data_flow, name="Test Flow", version=1, work_queue_name="iron_horse", storage=GitHub( repository='https://github.com/xxxxx/xxxxxx.git', access_token=Secret.load("block-type/block-name").get(), reference='master' ), I also checked the slug name using command "> prefect block ls" and I pasted the values into my script to make sure that I did not misspell the slug. What am I doing wrong?

Hello, I updated Server version to 2.10.10 to address the issue that I reported here. Unfortunately some other issue started to come through. My flows are now run 2 time in a row one by one. 1. Prefect 2.10.10 on server & agent(not worker) 2. Run on k8s-job 3. There is only one agent here 4. Flow gets completed and then goes to running again after 5 seconds. a. We limit concurrency on this queue to 40 and we usually get more than 40 flows running so they get late(might this be a hint?)

Hi, If a flow is crashed due to worker termination, how it can be automatically reschedule if worker is started again?

Hi, In Prefect 1.0 we are using prefect.tasks.postgres.postgres to perform dtabase operations on Postgres database. In Prefect 2.0, prefect.tasks.postgres.postgres are no longer supported, after some reseach I found two ways to do Postgres operations prefect-postgres and prefect-sqlalchemy. Can anybody please guide which one I should use?

Hi, Scenario to a possible bug: Step 1: Start Prefect Server and worker. Step 2: Run a long running deployment let say

Deployment-1

. Step 3: Prefect Server is killed/stopped. Step 4: After some time, restarted Prefect Server. So the flow run for

Deployment-1

stuck in running status and its neither

CRASH

or

CANCEL

. So how to handle such scenarios??

I have started using prefect-sqlalchemy but getting following error while establishing connection with Postgres using following command

def setdatabaseconnection():
    connector = SqlAlchemyConnector(
        connection_info=ConnectionComponents(
            driver=SyncDriver.POSTGRESQL_PSYCOPG2,
            username="postgres",
            password="",
            host="localhost",
            port=5432,
            database="claimshero_dev",
        )
    )

    connector.save("CH_DATABASE_CONNECTION")

Error : pydantic.error_wrappers.ValidationError: 2 validation errors for BlockDocumentCreate name name must only contain lowercase letters, numbers, and dashes (type=value_error) root Names must be provided for block documents. (type=value_error)

Hi, Is there is any way that task cache can be store on S3 instead Local storage?

Hi everyone! How can I create a variable (this one) using python package prefect 2.10.10?

Hey team ! I’m trying to run my flows in ECS Tasks but I’m constantly facing the same issue: It seems tasks are running the container image and not actually pointing to the flow entrypoint and path. I’m using a ECSTask block and defining my custom image in the

containerDefinition

inside my

task_definition

field. I created the Deployment from:

deployment = Deployment.build_from_flow(
    flow=template_local_flow,
    name="staging",
    infrastructure=ECSTask.load("staging"),
    work_queue_name="staging",
    path="/opt/prefect/flows",
    output='staging-deployment.yaml'
)

My docker image looks like this:

FROM prefecthq/prefect:2-python3.10
ENV DEBIAN_FRONTEND=noninteractive

RUN apt-get update && \
    apt-get dist-upgrade -y && \
    apt-get autoremove -y

COPY --from=golang:latest /usr/local/go/ /usr/local/go/

ENV PATH="/usr/local/go/bin:${PATH}"

RUN apt-get install -y python3 python3-dev python3-pip git

RUN cd /opt && \
    git clone <https://github.com/zmap/zdns> && \
    cd zdns && \
    go build && \
    ln -s /opt/zdns/zdns /usr/local/bin

RUN mkdir -p /opt/prefect/flows

ADD . /opt/prefect/flows

RUN pip3 install -r /opt/prefect/flows/requirements.txt

WORKDIR /opt/prefect/flows

ENV PYTHONPATH "${PYTHONPATH}:/opt/prefect/flows"

My

deployment.yaml

specifies:

storage: null
path: /opt/prefect/flows
entrypoint: flow.py:template_local_flow

But when I run and trigger a flow in the UI, all my container outputs is the

PREFECT

message. Can anyone help please?

Hello everyone, any idea on how to specify a runtime when using a docker worker ? I'd like to run docker containers with the

--runtime nvidia

but I don't see that option anywhere ?

Hello, currently trying to configure a prefect flow schedule but the behavior doesn't seem to be exactly what I expected. I'm trying to use a RRule string of

FREQ=HOURLY;INTERVAL=1;WKST=MO;BYDAY=MO,TU,WE,TH,FR;BYMINUTE=30

but instead of generating one flow each for 9:30, 10:30, 11:30 etc it seems like it creates multiple flows for these hours. Was wondering if there was a way I could solve this without having to manually adjust the

PREFECT_API_SERVICES_SCHEDULER_MIN_RUNS

value

Hi Prefect community, I'm a DevOps engineer, and we are currently running a self-hosted Prefect server on an EKS cluster. I'm aware that the self-hosted version doesn't support authorization out of the box. However, I was wondering if any members have experience in adding an authentication layer for the Prefect UI, such as using oauth2-proxy. Thank you.

any idea what is this exception about in the agent?

16:12:02.611 | ERROR  | asyncio - Exception in callback SubprocessStreamProtocol.pipe_data_received(1, b'\x1b[0m16:1...m in 0.12s]\n')
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]: handle: <Handle SubprocessStreamProtocol.pipe_data_received(1, b'\x1b[0m16:1...m in 0.12s]\n')>
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]: Traceback (most recent call last):
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]:  File "/usr/lib/python3.9/asyncio/events.py", line 80, in _run
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]:   self._context.run(self._callback, *self._args)
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]:  File "/usr/lib/python3.9/asyncio/subprocess.py", line 73, in pipe_data_received
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]:   reader.feed_data(data)
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]:  File "/usr/lib/python3.9/asyncio/streams.py", line 472, in feed_data
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]:   assert not self._eof, 'feed_data after feed_eof'
May 23 16:12:02 ip-172-30-1-204 start_agent.sh[1481686]: AssertionError: feed_data after feed_eof

prefect 2.10.10

actually it is a bug https://github.com/PrefectHQ/prefect-dbt/issues/146. Any eta about that?

What’s up with not being able to use SqlAlchemy 2.0 with prefect 2.8?

hi there - I'm using Prefecet v1, how exactly do I get access to the exception that caused a Flow failure? I'm using state handlers and I've tried inspecting the

new_state

variable including

new_state.message

,

new_state.result

and even

new_state.___getstate___()

- however none of those state attributes gets me anything meaningful. All I can ever find is "'Some reference tasks failed.'"

One really helpful improvement suggestion, if I could be so bold: Any way to fix the size of the schedule and name columns? As the schedule and deployment names get longer, the switch to enable and disable the flow is moved off the screen, which makes it tougher to turn them off and on. This might be fixed in a future version, so please let me know if I should just update instead.

Hello, I'm trying to use prefect-sendgrid but when I

pip install prefect-sendgrid

, I ran to an issue. Any suggestions for that?

ERROR: Could not find a version that satisfies the requirement prefect-sendgrid (from versions: none)
ERROR: No matching distribution found for prefect-sendgrid

Hello all! 👋 I am really struggling to setup a deployment for my bitbucket repo. I have already setup a "BitBucket Credentials" block and a "BitBucket Repository" block and am creating my deployment as follows:

bitbucket_repository_block = BitBucketRepository.load("prefect-bitbucket-repo")

deployment = Deployment.build_from_flow(
    flow           = test_flow,
    name           = "test_deployment",
    work_pool_name = "default-agent-pool",
    storage        = bitbucket_repository_block,
    path           = "/tmp/prefect/",
    entrypoint     = "flows/test.py:test_flow",
    tags=["DEV"]
)

deployment.apply()

However when I deploy this and start a run, the agent generates this error:

12:48:47.460 | INFO    | Flow run 'cornflower-leopard' - Downloading flow code from storage at '/tmp/prefect/'
12:48:48.987 | ERROR   | Flow run 'cornflower-leopard' - Flow could not be retrieved from deployment.
...
...
distutils.errors.DistutilsFileError: cannot copy tree '/tmp/prefect': not a directory

Any ideas?