我有一個使用 Dask 編寫的大型數(shù)據(jù)提取作業(yè)，其中每個任務(wù)將從數(shù)十個數(shù)據(jù)庫的大量表中查詢一個表。對于每個數(shù)據(jù)庫實例，我想限制一次連接的任務(wù)數(shù)量（即限制）。例如，我可能有 100 個任務(wù)連接到數(shù)據(jù)庫 A，100 個任務(wù)連接到數(shù)據(jù)庫 B，100 個任務(wù)連接到數(shù)據(jù)庫 C，等等，并且我想確保在任何給定時間連接到任何數(shù)據(jù)庫的任務(wù)不超過 20 個。我發(fā)現(xiàn) Dask 提供了基于工作線程資源（CPU、MEM、GPU 等）的約束，但是數(shù)據(jù)庫資源是“全局”的，因此對于任何 Dask 工作線程來說都不是特定的。Dask 是否提供任何方法來對任務(wù)并發(fā)性的此類約束進行建模？
1 回答
幕布斯6054654
TA貢獻1876條經(jīng)驗 獲得超7個贊
Dask 提供分布式信號量，可以限制對數(shù)據(jù)庫等資源的并發(fā)訪問。
例子
import time
from dask.distributed import Client, Semaphore
client = Client(...)
def do_task(x, sem):
? ? with sem:
? ? ? ? time.sleep(5)
? ? ? ? return x
# allow no more than 5 tasks to run concurrently
sem = Semaphore(max_leases=5, name="Limiter")
# submit jobs that use the semaphore
futures = client.map(do_task, range(20), sem=sem)
# collect results
results = client.gather(futures)
添加回答
舉報
0/150
提交
取消