the concurrency limit shows 181 active tasks for the tag but when i try to find the tasks by id ( after inspecting this tag ) these tasks are not found.. how do i fix this ?

Getting this error while running flows from prefect cloud:

11:13:27.904 | ERROR   | Flow run 'pink-pogona' - Crash detected! Request to <prefect cloud url >/task_runs/ failed: Traceback (most recent call last):
File "/home/<my_username>/<my_venv_name>/lib/python3.8/site-packages/anyio/streams/tls.py", line 108, in _call_sslobject_method
    result = func(*args)
File "/usr/lib/python3.8/ssl.py", line 944, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLSyscallError: Some I/O error occurred (_ssl.c:1131)

#prefect-community #prefect-cloud

Hey folks. Question for prefect 2.0 cloud: I recently (starting with 12th of November) got some individual flow runs in state "late" - however I did not add or change pipelines. They were picked up reasonably fast after entering late, but I'm wondering, is there anything I can do to debug why they were late? Alternative question: Is there a setting I can adjust which determines when a flow counts as "late"? My work queues are all with "concurrency: unlimited" and no task concurrency is used.

And an additional educational question 🤓 If I use a a block for infrastructure (like kubernetes-job) - then execute ``prefect deployment build`` using this block - does the deployment somehow have a link to this block? Meaning, if I change something in the block after creation of a deployment, is this reflected in next flow run of this deployment?

Hi. How to call sync task in async flow?(Prefect 2)

Hey! Does prefect have python method for this command

prefect work-queue set-concurrency-limit

?

hey Prefect. Do you have any plans on building a terraform provider or any IAC tool to manage prefect v2?

Hey guys if i have a flow scheduled to run and new version is registered the flow will be archived right?

Hey Prefect team! Is there a way to transfer ownership of a workspace?

If I have got envFrom : [] in my k8 job template should using infra_overrides on a deploy override it - as I cannot get it to work

Come watch our amazing @Taylor Curran Getting Started Demo

Hello! I have a flow that runs 10 tasks using a ray task runner. After these tasks all individually finish, the flow crashes with the following error:

Crash detected! Request to <https://api.prefect.cloud/api/accounts/CENSORED/workspaces/CENSORED/flow_runs/CENSORED/set_state> failed: Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/httpcore/backends/asyncio.py", line 33, in read
    return await self._stream.receive(max_bytes=max_bytes)
  File "/usr/local/lib/python3.8/site-packages/anyio/streams/tls.py", line 195, in receive
    data = await self._call_sslobject_method(self._ssl_object.read, max_bytes)
  File "/usr/local/lib/python3.8/site-packages/anyio/streams/tls.py", line 137, in _call_sslobject_method
    data = await self.transport_stream.receive()
  File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 1265, in receive
    await self._protocol.read_event.wait()
  File "/usr/local/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError

The URL points to the flow's own id. What's going on here? Did anyone else experience this?

Hi all, I’m trying to run prefect flows in a restricted/corporate K8s. I cannot rely on KubernetesJob as it triggers new jobs and ressource creations is blocked outside of prod deployment. Is there a way to run flows inside of a pod waiting to be triggered by an agent. Would https://docs.prefect.io/api-ref/prefect/infrastructure/#prefect.infrastructure.process be the solution ? thanks

I am using prefect 2.6.6 local server:I am using run_deployment() to trigger flow_run from deployment A and B. Second flow should start once the first is finished. whenever I trigger the flow its failing with below error. Can someone please help me with this.Thanks. Encountered exception during execution: Traceback (most recent call last): File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/client/orion.py", line 1351, in read_deployment_by_name response = await self._client.get(f"/deployments/name/{name}") File "/home/x/anaconda3/lib/python3.9/site-packages/httpx/_client.py", line 1751, in get return await self.request( File "/home/x/anaconda3/lib/python3.9/site-packages/httpx/_client.py", line 1527, in request return await self.send(request, auth=auth, follow_redirects=follow_redirects) File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/client/base.py", line 182, in send response.raise_for_status() File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/client/base.py", line 125, in raise_for_status raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url 'https://api.prefect.cloud/api/accounts/e25149ff-b6ff-4b7b-8d35-15fedaa1a20b/workspaces/fb6e8c45-2229-4aaf-8c98-9c03a6f2ae86/deployments/name/db2_replace' Response: {'detail': 'Not Found'} For more information check: https://httpstatuses.com/404 The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/engine.py", line 610, in orchestrate_flow_run result = await run_sync(flow_call) File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 68, in run_sync_in_worker_thread return await anyio.to_thread.run_sync(call, cancellable=True) File "/home/x/anaconda3/lib/python3.9/site-packages/anyio/to_thread.py", line 28, in run_sync return await get_asynclib().run_sync_in_worker_thread(func, *args, cancellable=cancellable, File "/home/x/anaconda3/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 818, in run_sync_in_worker_thread return await future File "/home/x/anaconda3/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 754, in run result = context.run(func, *args) File "/home/x/Desktop/mlp-prefect/mlp_prefect/flows/standard/pipeline/mainflow_db2_etl.py", line 13, in mainflow db2_load_flow_run = run_deployment(name="db2_replace") File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 197, in coroutine_wrapper return run_async_from_worker_thread(async_fn, *args, **kwargs) File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/utilities/asyncutils.py", line 148, in run_async_from_worker_thread return anyio.from_thread.run(call) File "/home/x/anaconda3/lib/python3.9/site-packages/anyio/from_thread.py", line 35, in run return asynclib.run_async_from_thread(func, *args) File "/home/x/anaconda3/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 847, in run_async_from_thread return f.result() File "/home/x/anaconda3/lib/python3.9/concurrent/futures/_base.py", line 446, in result return self.__get_result() File "/home/x/anaconda3/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result raise self._exception File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/client/utilities.py", line 47, in with_injected_client return await fn(*args, **kwargs) File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/deployments.py", line 78, in run_deployment deployment = await client.read_deployment_by_name(name) File "/home/x/anaconda3/lib/python3.9/site-packages/prefect/client/orion.py", line 1354, in read_deployment_by_name raise prefect.exceptions.ObjectNotFound(http_exc=e) from e prefect.exceptions.ObjectNotFound

Hi, I am getting this error importing fivetran's tasks:

from prefect_fivetran.connectors import fivetran_sync_flow
ImportError: cannot import name 'fivetran_sync_flow' from 'prefect_fivetran.connectors' (/usr/local/lib/python3.9/site-packages/prefect_fivetran/connectors.py)

Hello Prefect community, I was curious to get people's opinion about Fivetran vs Airbyte vs Stitch or any other connectors. We want to be able to pull data from Salesforce, DWH (Looker, Snowflake, BQ, Redshift), Pendo, Heap and dump into BQ (our DWH). Read a bunch of blog posts, but was hoping to hear more personal experiences.

Has anyone figured out a stable way to run a flow constantly? We have tried using

while true:

in the flow, but that has always shut down in the past after a few hours on 2.0

If I have a flow running that grabs all other flow runs with read_flow_runs, and it picks out the flow run that it is supposed to follow. How do I make it wait until the other flow is done before continuing? Can I do this without a busy wait?

If this isn't clear:

flow run A is running
flow run A has list of info on flow runs B -> G
flow run A picks out flow run D
I want flow run A to wait until flow run D is done before continuing, how do I do this step?

Hey all, I was wondering if anyone has run into the following error while trying to run a prefect flow (version 1) using the pandas package

Segmentation fault (core dumped)

. I have run this using different version of pandas (1.5.1 and 1.3.5) and still get the same error. I am able to import pandas and run a command such as

pd.show_versions()

without issues but whenever I try to run anything that would create a dataframe I get the segfault error. Including my currently installed versions and the flow I am trying to run below. Thanks in advance!

Open Question, asking to learn: Why would I schedule an AWS Batch job with Prefect, instead of using the native Batch scheduling?

Does anyone know of a Prefect native way (some context variable, etc.) to log out from a Prefect flow the k8s job name it ran on? It would be very useful for monitoring/debugging purposes.

What is the correct method to call a synchronous task from an asynchronous flow in Prefect 2? When trying to this with the attached code, our asynchronous `async_task`’s will execute, but our

sync_task

will never start and the flow just seems to hang. The intention here is to run some asynchronous tasks, gather the results, then send those results to a synchronous or blocking task that can be executed sequentially (or submitted to a concurrent task runner). The equivalent code works in a standard Python script if we remove our Prefect decorators.

from prefect import flow, task
import asyncio, time

@task
async def async_task(n):
    await asyncio.sleep(1)
    return n

@task
def sync_task(res):
    time.sleep(1)
    print(res)

@flow
async def async_flow():
    res = await asyncio.gather(*(async_task(n) for n in range(0, 3, 1)))
    sync_task(res)


if __name__ == "__main__":
    asyncio.run(async_flow())

See output in replies.

Getting this error with the fivetran syncs when I move

await

from the function, any ideas?

A 'sync_compatible' method was called from a context that was previously async but is now sync. The sync call must be changed to run in a worker thread to support sending the coroutine for 'load' to the main thread.

Hi Everyone, Had the same issue with k8 job deployments that was first found here

Failed to submit flow run to infrastructure

https://prefect-community.slack.com/archives/CL09KU1K7/p1663166587656539 Details in thread

Hi Prefect! I'm running a parent flow that maps the parameters into deployments runs. I have to wait that all deployments runs has been complete to execute other flow. My parent flow execute all the deployments and them runs correctly, but the flow after the deployments runs was not executed. I got

Invalid input ConnectionInputs.RECV_PING in state ConnectionState.CLOSED

complete logs in the thread.

Hey Prefect - I keep getting these errors when a flow is scheduled with my ECSTask block.

State message: Submission failed. RuntimeError: Timed out after 120.81494164466858s while watching task for status {until_status or 'STOPPED'}

Can I handle this somehow? is this handled by the

task_start_timeout_seconds

?

Hello, Question about prefect 1.0 free tier : • we have a long-running flow (about a week) that simply disappeared from existence, Thanos-style (it doesn’t appear anywhere, it’s not running, it’s not stopped, it’s not in history) • it says that free-tier 7 days of history, does that mean that flows that run for longer are simply “erased” like that?

Hi Team, I have my flows using private github storage block and ECS task infrasrtructure. I want to override the ECStask config to add ephemeralStorage for a particular flow during deployment. So how to do i achieve these custom ECS task config for flows which override the base config. Do I create a seperate block with the config or can i override it during deployment ?

Hi all, I’m struggling on using the radar and would need your guidance about it. My flows are running fine, but it seems that no radar is being displayed (see first picture), while I have a functional radar for a subflow being called (see second picture). The following flow relies on some data being passed downstream. Let me know how i can get a working Radar for it please. Thanks

@flow
def flow():

   f1 = subflow1()
   f2 = subflow2()

   data1 = task1.submit(wait_for=[f1, f2])
   data2 = task2().submit()
   data3 = task3.submit(data1, data2, wait_for=[data1, data2])

    if data3.result():
       data4 = task4.submit(data3, wait_for=[data3])
       task5.submit(data4, wait_for=[data4])
    else:
        pass

#prefect-community #prefect-cloud Getting this error while running flows from prefect cloud

12:41:52.823 | ERROR   | Task run 'flow-3aa4322b-126' - Crash detected! Execution was cancelled by the runtime environment
.                                                                                                                         
12:46:59.458 | ERROR   | Flow run 'eminent-heron' - Crash detected! Request to <https://api.prefect.cloud/api/accounts/2f48>
cb6f-0049-4569-879d-7123d9113e31/workspaces/9da885d9-9d9f-4ee9-9b42-84bfb4a3d9aa/task_runs/ failed: Traceback (most recent
 call last):                                                                                                              
  File "/home/shrutihande/python_venv/lib/python3.8/site-packages/httpcore/backends/asyncio.py", line 33, in read         
    return await self._stream.receive(max_bytes=max_bytes)                                                                
  File "/home/shrutihande/python_venv/lib/python3.8/site-packages/anyio/streams/tls.py", line 171, in receive             
    data = await self._call_sslobject_method(self._ssl_object.read, max_bytes)                                            
  File "/home/shrutihande/python_venv/lib/python3.8/site-packages/anyio/streams/tls.py", line 115, in _call_sslobject_meth
od                                                                                                                        
    data = await self.transport_stream.receive()                                                                          
  File "/home/shrutihande/python_venv/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 1105, in receive     
    await self._protocol.read_event.wait()                                                                                
  File "/usr/lib/python3.8/asyncio/locks.py", line 309, in wait
    await fut
asyncio.exceptions.CancelledError