Usando `dask` para llenar `boost_histograms` almacenados en clase en paralelo

Tengo una pregunta dask. boost_histogramTengo una estructura de código de la siguiente manera:

Tengo una clase definida en algún script:

class MyHist:
 def __init__(....):
 self.bh = None
 def make_hist(...):
 axis = bh.axis.Regular(....)
 @dask.delayed
 def fill_hist(data)
 self.bh.fill(data)

y en otro script quiero llenar varios histogramas en paralelo con dask. Los datos son awkwardmatrices que leo de la entrada, y para eso hago algo como:

 from dask.distributed import Client
 cl = Client()
 histos = [MyHist(..), MyHist(another...)]
 for i, file in enumerate(files):
 data = dask.delayed(open_file(file))
 for myhist in histos:
 if i ==0: myhist.make_hist()
 fill_results.append(dask.delayed(myhist.fill_hist(data))
 dask.compute(*fill_results)

Si luego trato de llamar

for j, h in enumerate(histos):
 print(h.bh) 

Obtengo histogramas vacíos. Sin embargo, si imprimo el histograma de impulso dentro de la fill_histfunción, los histogramas parecen estar llenos.

¿El dask cálculo crea una copia profunda o algo del MyHistobjeto para realizar el cálculo y, por lo tanto, llena el bhasociado con esa copia? o estoy haciendo algo mal aquí?

Solución del problema

Creo que la definición de su clase dificulta trabajar correctamente con dask; es decir, probablemente le resulte más fácil si su fill_histmétodo fuera realmente una función libre. Y en su ciclo, en realidad está llamando dask.delayeda un delayedmétodo ya (es probable que esto no sea lo que quiere hacer):

fill_results.append(dask.delayed(myhist.fill_hist(data))
#   ^^^^^^^^^
#   already delayed method

Mi sugerencia sería ir con una función gratuita:

@dask.delayed
def fill_hist(data, axes):
 h = bh.Histogram(*axes)
 h.fill(data)
 return h
@dask.delayed
def open_file(fname):
 data = some_function_to_get_data(fname)
 return data
axes = (bh.axis.Regular(100, -10, 10),) # tuple with a single axis
tasks = []
for f in files:
 data = open_file(f)
 hist = fill_hist(data=data, axes=axes)
 tasks.append(hist)
results = dask.compute(tasks)

¡ Este patrón es muy similar a cómo dask-histogramfunciona, que tiene soporte para dask-awkward!