Store partial results in dask delayed iterations

Viewed 601

I have a current long iteration process where I run a calculation and every x iteration I store the results to DB.

For example, iterate fun() function over range(20) and save every 5 results with save_results:

import time

def fun(x):
    time.sleep(0.1*x)
    return(0.1*x)

def save_results(result):
    # originally stores the new data to DB
    print result  # print as example

result = []
for i in range(20):
    result.append(fun(i))
    if i%5==4:
        save_results(result[-5:])

I want to parallelize the process with dask delayed and compute methods. But if I run it like in the following example, the store_results occurs before the compute:

import dask as da

result = []
for i in range(20):
    result.append(da.delayed(fun)(i))
    if i%5==4:
        save_results(result[-5:])

result = da.compute(result)[0]

and therefore, instead of storing the results every 5 iterations, I store a list of delayed objects:

[Delayed('fun-202f7e28-e594-4926-a5cd-5931dbc99d6b'),
Delayed('fun-d2bf2bc9-a4f3-46d7-adb7-84114a68b482'),
Delayed('fun-c34f2c04-3e25-47fa-8165-1ee7c786aaf6'),
Delayed('fun-a4edd3fc-442d-4ec1-8a0e-320bd9315a61'),
Delayed('fun-c7b48e2c-cb66-472e-85c5-fe6c595fa1ec')]

How can I overcome the issue, and store every 5 new results to DB?

1 Answers
Related