Issues when using DaskTaskRunner at scale - running out of available connections when connecting to Orion DB across distributed workers

Question

Hey guys I m running a flow with a DaskTaskRunner that spawns a task that takes around 20seconds and apparently it s running sequentially My machine has 16 threads and I see in the UI each task only being run after another one finishes Any inputs ```23 04 54 057 | INFO | prefect engine Created flow run optimal snail for flow test1 23 04 54 057 | INFO | prefect task runner dask Creating a new Dask cluster with `distributed deploy local LocalCluster` 23 04 55 980 | INFO | prefect task runner dask The Dask dashboard is available at <http 127 0 0 1 8787 status> 23 05 00 371 | INFO | Flow run optimal snail Created task run Execute values of the query 8165e3c8 0 for task Execute values of the query 23 05 00 372 | INFO | Flow run optimal snail Executing Execute values of the query 8165e3c8 0 immediately 23 05 24 612 | INFO | Task run Execute values of the query 8165e3c8 0 Finished in state Completed 23 05 26 370 | INFO | Flow run optimal snail Created task run Execute values of the query 8165e3c8 1 for task Execute values of the query 23 05 26 370 | INFO | Flow run optimal snail Executing Execute values of the query 8165e3c8 1 immediately 23 05 42 533 | INFO | Task run Execute values of the query 8165e3c8 1 Finished in state Completed 23 05 44 295 | INFO | Flow run optimal snail Created task run Execute values of the query 8165e3c8 2 for task Execute values of the query 23 05 44 296 | INFO | Flow run optimal snail Executing Execute values of the query 8165e3c8 2 immediately 23 06 09 538 | INFO | Task run Execute values of the query 8165e3c8 2 Finished in state Completed ``` As you see in these logs it s not really spawning more tasks I was under the impression that the call to a function with the ` task` decorator would be non blocking if made within a flow ` flow` which uses DaskTaskRunner

Zanie · Answer

Hey Jean wave you ll want to use ` submit` for those task calls to run them in the background

Jean · Answer

Thanks < Zanie> that s working got other issues now tho

Jean · Answer

Apparently the amount of data I m sending to each task is too much i e a list of 10k records 20 values

Jean · Answer

Getting a bunch of errors from sqlite ValueError no active connection sqlalchemy exc OperationalError sqlite3 OperationalError database is locked

Zanie · Answer

If you re performing a bunch of concurrency we recommend switching to Postgres as a backing database

Jean · Answer

I m getting sqlalchemy exc InvalidRequestError The asyncio extension requires an async driver to be used The loaded psycopg2 is not async < Zanie> when I try to use POSTGRES

Jean · Answer

I have both asyncpg and psycopg2 installed

Zanie · Answer

There s an issue with the rendering of settings in our documentation reference right now sounds like your connection string is wrong <https github com PrefectHQ prefect blob main src prefect settings py L438 L456>

Zanie · Answer

See also <https github com PrefectHQ prefect issues 6548>

Jean · Answer

That did the trick thanks

Jean · Answer

So Dask is complaining to use Scatter and workers are crashing

Jean · Answer

```23 52 10 536 | INFO | Flow run obedient seal Created task run Execute values of the query 8165e3c8 0 for task Execute values of the query home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages distributed worker py 2936 UserWarning Large object of size 2 75 MiB detected in task graph task lt prefect tasks Task object at 0x7fb7a4f1a ENABLED=True Consider scattering large objects ahead of time with client scatter to reduce scheduler burden and keep data on workers future = client submit func big data bad big future = client scatter big data good future = client submit func big future good warnings warn 23 52 11 735 | INFO | Flow run obedient seal Submitted task run Execute values of the query 8165e3c8 0 for execution 23 52 15 432 | INFO | Flow run obedient seal Created task run Execute values of the query 8165e3c8 1 for task Execute values of the query 23 52 16 169 | INFO | Flow run obedient seal Submitted task run Execute values of the query 8165e3c8 1 for execution 23 52 21 062 | INFO | Flow run obedient seal Created task run Execute values of the query 8165e3c8 2 for task Execute values of the query 23 52 22 476 | INFO | Flow run obedient seal Submitted task run Execute values of the query 8165e3c8 2 for execution 23 52 24 879 | INFO | Flow run obedient seal Created task run Execute values of the query 8165e3c8 4 for task Execute values of the query 23 52 26 272 | INFO | Flow run obedient seal Submitted task run Execute values of the query 8165e3c8 4 for execution 23 52 27 541 | ERROR | prefect orion Encountered exception in request Traceback most recent call last File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages starlette middleware errors py line 162 in call await scope receive send File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages starlette middleware exceptions py line 75 in call raise exc File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages starlette middleware exceptions py line 64 in call await scope receive sender File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages fastapi middleware asyncexitstack py line 21 in call raise e File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages fastapi middleware asyncexitstack py line 18 in call await scope receive send File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages starlette routing py line 680 in call await route handle scope receive send File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages starlette routing py line 275 in handle await scope receive send File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages starlette routing py line 65 in app response = await func request File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages prefect orion utilities server py line 101 in handle response scoped depends response = await default handler request File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages fastapi routing py line 231 in app raw response = await run endpoint function File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages fastapi routing py line 160 in run endpoint function return await dependant call values File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages prefect orion api task runs py line 49 in create task run model = await models task runs create task run File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages prefect orion database dependencies py line 117 in async wrapper return await fn args kwargs File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages prefect orion models task runs py line 61 in create task run await session execute insert stmt File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy ext asyncio session py line 215 in execute result = await greenlet spawn File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy util concurrency py3k py line 126 in greenlet spawn result = context throw sys exc info File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy orm session py line 1711 in execute conn = self connection for bind bind File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy orm session py line 1552 in connection for bind return self transaction connection for bind File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy orm session py line 747 in connection for bind conn = bind connect File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy future engine py line 406 in connect return super Engine self connect File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy engine base py line 3315 in connect return self connection cls self close with result=close with result File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy engine base py line 96 in init else engine raw connection File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy engine base py line 3394 in raw connection return self wrap pool connect self pool connect connection File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy engine base py line 3361 in wrap pool connect return fn File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool base py line 320 in connect return ConnectionFairy checkout self File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool base py line 884 in checkout fairy = ConnectionRecord checkout pool File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool base py line 486 in checkout rec = pool do get File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool impl py line 145 in do get with util safe reraise File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy util langhelpers py line 70 in exit compat raise File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy util compat py line 208 in raise raise exception File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool impl py line 143 in do get return self create connection File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool base py line 266 in create connection return ConnectionRecord self File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool base py line 381 in init self connect File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy pool base py line 691 in connect exec w sync on first run self dbapi connection self File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy event attr py line 326 in exec w sync on first run self args kw File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy event attr py line 334 in call fn args kw File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy engine create py line 658 in on connect do on connect dbapi connection File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy dialects postgresql asyncpg py line 1109 in connect conn await self setup asyncpg jsonb codec conn File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy util concurrency py3k py line 68 in await only return current driver switch awaitable File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy util concurrency py3k py line 121 in greenlet spawn value = await result File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages sqlalchemy dialects postgresql asyncpg py line 1086 in setup asyncpg jsonb codec await asyncpg connection set type codec File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages asyncpg connection py line 1222 in set type codec typeinfo = await self introspect type typename schema File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages asyncpg connection py line 473 in introspect type rows = await self execute File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages asyncpg connection py line 1659 in execute result = await self execute File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages asyncpg connection py line 1684 in execute return await self do execute File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages asyncpg connection py line 1734 in do execute result = await executor stmt timeout File asyncpg protocol protocol pyx line 182 in bind execute File asyncpg protocol protocol pyx line 696 in asyncpg protocol protocol BaseProtocol get timeout impl asyncio exceptions TimeoutError 2022 09 28 23 52 29 034 distributed client ERROR Traceback most recent call last File home ubuntu cache pypoetry virtualenvs fDAH7s3U py3 10 lib python3 10 site packages distributed client py line 1579 in close await self scheduler comm close asyncio exceptions CancelledError 23 52 29 314 | ERROR | Flow run obedient seal Crash detected Execution was interrupted by an unexpected exception ```

Zanie · Answer

I don t have a great way to help with that warning yet you may need to manually scatter large data types to the Dask cluster which would require something like <https github com PrefectHQ prefect dask pull 33>

Zanie · Answer

The failure here though is due to the database running out of available connections

Jean · Answer

Which database Postgres The one Orion is using

Zanie · Answer

Yeah it looks like it Basically you re beyond a simple flow and you need to run your server separately with `prefect orion start`

Jean · Answer

I m running my server with prefect orion start

Zanie · Answer

Did you set `PREFECT API URL` before running the flow I can see a call directly to the database in your flow run s traceback which indicates it s not making requests to your separate server

Jean · Answer

It s not even a webservice just a screep with a flow that iterates a 1m list and each iteration sends a task of 20s

Jean · Answer

Hmmmm let me check

Jean · Answer

Yeah I didnt set prefect api url in my script but I can see the flows tasks running in localhost 4200

Zanie · Answer

I m tempted to make `PREFECT API URL` auto detect a locally running API to improve this kind of ux Otherwise what s happening here is the run and your standalone server are sharing a database so it looks like it s working but really each task run is standing up a copy of the entire Orion API in memory and it connects to the database directly

Zanie · Answer

You ll need to set `PREFECT API URL` to have the task runs send API requests to your centralized server which then holds a single roughly connection to the database

Jean · Answer

ok just did `prefect config set PREFECT API URL=<http 127 0 0 1 4200 api>`

Jean · Answer

That should be it or should I do something else in the python script

Zanie · Answer

That should do it

Jean · Answer

It s very confusing since I can see the flows tasks inside the UI but it s not connected to it Not sure I understand

Zanie · Answer

Let s see here `prefect orion start` starts the Orion API and connects to the database The UI is hosted as well and asks the API for data to display

Zanie · Answer

When you run a flow without an API URL set It starts the Orion API in memory which connects to the same database as your standalone server

Zanie · Answer

The UI asks the standalone server for data and it reads the flow run from database

Jean · Answer

Ah I see okay

Jean · Answer

Well not it has run a bit but still failing at one point

Zanie · Answer

However because your flow has tasks that are happening on Dask workers we end up making a bunch of copies of the Orion API which each requires a connection to the database etc

Zanie · Answer

What s it fail with now disappointed

Zanie · Answer

If you set `PREFECT DEBUG MODE=1` we ll get the traceback for the crash

Jean · Answer

RuntimeError The connection pool was closed while 2 HTTP requests responses were still in flight 2022 09 29 00 07 06 316 distributed process WARNING lt AsyncProcess Dask Worker process from Nanny gt process 9283 exit status was already read will report exitcode 255 sys 1 RuntimeWarning coroutine get task call return value was never awaited

Jean · Answer

raise mapped exc message from exc httpx ReadError

Jean · Answer

File home ubuntu cache pypoetry virtualenvs data fDAH7s3U py3 10 lib python3 10 site packages distributed comm tcp py line 225 in read frames nbytes = await stream read bytes fmt size asyncio exceptions CancelledError

Zanie · Answer

Ah okay now we ve reached a real bug

Zanie · Answer

It looks like Dask may have crashed I d need the full traceback to tell though since once the flow crashes we shut everything down

Jean · Answer

out txt

Jean · Answer

I mean maybe I m sending too much data If I do it manually it works fine or if I reduce the bach size currently at 7500 at 2500 it works fine I think

Zanie · Answer

I ve got to head out for the night Thanks for working through this with me though We re aware of these `httpx ReadError` s when running tasks at scale and we re putting together a team to fix them

Zanie · Answer

If you can put together a minimal example that d be really helpful

Zanie · Answer

How many task runs are you running The data is being passed as a parameter to your tasks Are you passing large amounts of data elsewhere

Jean · Answer

< Zanie> ``` usr bin env python3 from random import choice from typing import Any Union import psycopg2 from psycopg2 extras import execute values from prefect import flow task from prefect dask task runners import DaskTaskRunner redshift host = redshift database = redshift user = redshift pass = def connect to redshift host str = redshift host db str = redshift database user str = redshift user pw str = redshift pass conn = psycopg2 connect host=host database=db user=user password=pw port=5439 return conn task name= Execute values of the query retries=5 retry delay seconds=10 def batch insert into redshift batch list dict str Union str list Any conn = connect to redshift cursor = conn cursor execute values cursor f INSERT INTO batch table join batch fields VALUES s batch values conn commit conn close flow name= Sending batch of data to Redshift task runner=DaskTaskRunner def batch insert records batch size = 3 500 total b = 1 918 573 b fields = i for i in range 20 b possible values = 100 12312312 sdsdsds 12312321 sssss 2022 05 03T04 02 02 b = 0 while b lt total b chunk = batch size if b + batch size gt total b chunk = total b b b += chunk b batch dict str Union str list Any = table b fields field for field in b fields values for in range chunk b batch values append choice b possible values for in range 20 batch insert into redshift submit b batch if name == main batch insert records ```

Jean · Answer

How many task runs are you running In the example above it would be 549 1 918 573 3 500 The data is being passed as a parameter to your tasks Yes Are you passing large amounts of data elsewhere I don t think so only to the task inside the flow

Anna Geller · Answer

< Jean> since your flow is doing a bunch of IO operations have you tried using `ConcurrentTaskRunner` instead of `DaskTaskRunner` You may end up getting better performance and fixing this issue at the same time

Zanie · Answer

Thanks for the additional details Anna has a good point that all we can run the tasks concurrently with threads All this data will be passed in memory and we ll also do better job sharing clients between tasks You may see an improvement in performance there

Jean · Answer

< Anna Geller> not crashing anymore but performance is not better at all Tasks where being completed before crashing with dask around 130 tasks mark it s been 1 hour and only like 10 tasks have finished

Jean · Answer

I ll try running again each task supposed to finish in at most 30sec

Jean · Answer

And how concurrent the `ConcurrentTaskRunner` actually is

Jean · Answer

Yeah it s just getting stuck again somewhere < Zanie>

Zanie · Answer

gt And how concurrent the `ConcurrentTaskRunner` actually is It uses threading instead of multiprocessing

Zanie · Answer

This actually might be affected by <https github com PrefectHQ prefect pull 6948>

Zanie · Answer

Which I believe we will be releasing today

Jean · Answer

Ah okay let me know and I will test

Jean · Answer

As of now none of the approaches are working for me confused Concurrent Dask

Zanie · Answer

We have released that change now

Zanie · Answer

Let me know if it works

Jean · Answer

Sure