title: Load data from dask area in parellel in chunks
body: <p>I have a large xarray dataset loaded in a dask dataframe which contains data over both quite a large spatial and temporal range. What I'm trying to do is load this data into memory using dask by splitting it into smaller chunks and loading these in in parallel. Below is an example code of what I'm trying to do:</p>
  
  <pre><code>import numpy as np
  import xarray as xr
  
  def chunk(ds,x_ends,y_ends):
      '''
      Function which takes a large dataset which has been lazily loaded and specified
      indices within the dataset, and cuts out the chunk and loads it into memory. 
      '''
  
      chunk = ds.isel(x=slice(x_ends[0],x_ends[1]),y=slice(y_vals[0],y_vals[1]))
  
      with ProgressBar():
          chunk = chunk.compute()
      return chunk
  
  dval = np.random.randint(5,size=[10,100,100])
  x = np.linspace(0,100,101,dtype=int)
  y = np.linspace(0,100,101,dtype=int)
  time = np.linspace(0,10,11,dtype=int)
  
  data = xr.DataArray(dval,coords=[time,x,y],dims=['time','x','y'])
  
  x_vals = np.arange(0,len(data.x),1000)
  x_vals = np.append(x_vals,len(data.x))
  y_vals = np.arange(0,len(data.y),1000)
  y_vals = np.append(y_vals,len(data.y))
  
  for i in range(len(x_vals)-1):
      for j in range(len(y_vals)-1):
          chunk(data,[x_vals[i],x_vals[i+1]],[y_vals[j],y_vals[j+1]])
  </code></pre>
  
  <p>This does what I want it to do, but obviously not in parallel and isn't great with the double for loop. This will slot into a larger function where the data will have other operations applied on it. I also realise that the DataArray in the example isn't a dasked array.</p>
  
  <p>I previously tried using the dask.distributed Client class, but that broke the .compute() function. I feel like there is probably a relatively simple answer which I'm just looking in the wrong place for.</p>