我正在构建一个FastAPI应用程序,该程序将提供Dask数组的大部分内容。我想在FastAPI's asynchronous functionality旁边使用Dask-distributed's ability to operate asynchronously。下面是一个mcve,它演示了我正在尝试在应用程序的服务器端和客户端上执行的操作:
服务器端:
import time
import dask.array as da
import numpy as np
import uvicorn
from dask.distributed import Client
from fastapi import FastAPI
app = FastAPI()
# create a dask array that we can serve
data = da.from_array(np.arange(0, 1e6, dtype=np.int), chunks=100)
async def _get_block(block_id):
"""return one block of the dask array as a list"""
block_data = data.blocks[block_id].compute()
return block_data.tolist()
@app.get("/")
async def get_root():
time.sleep(1)
return {"Hello": "World"}
@app.get("/{block_id}")
async def get_block(block_id: int):
time.sleep(1) # so we can test concurrency
my_list = await _get_block(block_id)
return {"block": my_list}
if __name__ == "__main__":
client = Client(n_workers=2)
print(client)
print(client.cluster.dashboard_link)
uvicorn.run(app, host="0.0.0.0", port=9000, log_level="debug")
客户端
import dask
import requests
from dask.distributed import Client
client = Client()
responses = [
dask.delayed(requests.get, pure=False)(f"http://127.0.0.1:9000/{i}") for i in range(10)
]
dask.compute(responses)
在此设置中,compute()
中的_get_block
调用是“阻塞”的,一次仅计算一个块。我尝试了Client(asynchronous=True)
和client.compute(dask.compute(responses)
的各种组合,但没有任何改善。是否可以await
计算dask数组?
此行
block_data = data.blocks[block_id].compute()
是阻止呼叫。如果您改用client.compute(data.blocks[block_id])
,则可以得到一个可以与IOLoop结合使用的未来,只要Dask使用相同的循环即可。
请注意,Intake服务器非常希望以这种方式工作(它也渴望为数组和其他数据类型按块流传输数据)。