으로 반복 순차적으로 '비 스레드 저장소에 <code>dask.bag</code> 매우 큰의 요소를 제출해야

Question

는, 즉, 나는 가방이 큰에 때문에 나는 compute을 사용할 수 없습니다

for x in dbag: 
    store.add(x) 

같은 뭔가가 필요 기억에 맞게. 나는 distributed.as_completed과 같은 것을 더 필요로하지만, 그것은 distributed.as_completed이 아닌 가방에서 작동합니다.

Answer 1

아마 정상적인 계산을 계속 사용할 수 있지만, 당신은 당신의 처리의 종류에 따라 threading.Lock, 또는 multiprocessing.Lock을 만들 수 있습니다 잠금

def commit(x, lock=None): 
    with lock: 
     store.add(x) 

b.map(commit, lock=my_lock) 

을 추가 당신이 원하는 경우

을하고있는 것 as_completed를 사용하면 가방을 선물로 변환하고 as_completed를 사용할 수 있습니다.

from distributed.client import futures_of, as_completed 
b = b.persist() 
futures = futures_of(b) 

for future in as_completed(futures): 
    for x in future.result(): 
     store.add(x) 

또한 좀 더 현명

df = b.to_dataframe(...) 
for x in df.iteritems(...): 
    ... 

를 반복 않습니다 믿는 dataframe로 변환 할 수 있습니다