title: Best way to parallelize computation over dask blocks that do not return np arrays?
body: <p>I'd like to return a dask dataframe from an overlapping dask array computation, where each block's computation returns a pandas dataframe.  The example below shows one way to do this, simplified for demonstration purposes.  I've found a combination of <code>da.overlap.overlap</code> and <code>to_delayed().ravel()</code> as able to get the job done, if I pass in the relevant block key and chunk information.  </p>
  
  <p>Edit:
  Thanks to a @AnnaM who caught bugs in the original post and then made it general!   Building off of her comments, I'm including an updated version of the code.  Also, in responding to Anna's interest in memory usage, I verified that this does not seem to take up more memory than naively expected.</p>
  
  <pre><code>def extract_features_generalized(chunk, offsets, depth, columns):
      shape = np.asarray(chunk.shape)
      offsets = np.asarray(offsets)
      depth = np.asarray(depth)
      coordinates = np.stack(np.nonzero(chunk)).T     
      keep = ((coordinates &gt;= depth) &amp; (coordinates &lt; (shape - depth))).all(axis=1)    
      data = coordinates + offsets - depth
      df = pd.DataFrame(data=data, columns=columns)
      return df[keep]
  
  def my_overlap_generalized(data, chunksize, depth, columns, boundary):
      data = data.rechunk(chunksize)
      data_overlapping_chunks = da.overlap.overlap(data, depth=depth, boundary=boundary)
  
      dfs = []
      for block in data_overlapping_chunks.to_delayed().ravel():
          offsets = np.array(block.key[1:]) * np.array(data.chunksize)
          df_block = dask.delayed(extract_features_generalized)(block, offsets=offsets, 
                                                                depth=depth, columns=columns)
          dfs.append(df_block)
  
      return dd.from_delayed(dfs)
  
  data = np.zeros((2,4,8,16,16))  
  data[0,0,4,2,2] = 1
  data[0,1,4,6,2] = 1
  data[1,2,4,8,2] = 1
  data[0,3,4,2,2] = 1
  
  arr = da.from_array(data)
  df = my_overlap_generalized(arr, 
                              chunksize=(-1,-1,-1,8,8), 
                              depth=(0,0,0,2,2), 
                              columns=['r', 'c', 'z', 'y', 'x'],
                              boundary=tuple(['reflect']*5))
  df.compute().reset_index()
  </code></pre>
  
  <p>-- Remainder of original post, including original bugs --</p>
  
  <p>My example only does xy overlaps, but it's easy to generalize.  Is there anything below that is suboptimal or could be done better?  Is anything likely to break because it's relying on low-level information that could change (e.g. block key)?</p>
  
  <pre><code>def my_overlap(data, chunk_xy, depth_xy):
      data = data.rechunk((-1,-1,-1, chunk_xy, chunk_xy))
      data_overlapping_chunks = da.overlap.overlap(data, 
                                                   depth=(0,0,0,depth_xy,depth_xy), 
                                                   boundary={3: 'reflect', 4: 'reflect'})
  
      dfs = []
      for block in data_overlapping_chunks.to_delayed().ravel():
          offsets = np.array(block.key[1:]) * np.array(data.chunksize)
          df_block = dask.delayed(extract_features)(block, offsets=offsets, depth_xy=depth_xy)
          dfs.append(df_block)
  
      # All computation is delayed, so downstream comptutions need to know the format of the data.  If the meta
      # information is not specified, a single computation will be done (which could be expensive) at this point
      # to infer the metadata.
      # This empty dataframe has the index, column, and type information we expect in the computation.
      columns = ['r', 'c', 'z', 'y', 'x']
  
      # The dtypes are float64, except for a small number of columns
      df_meta = pd.DataFrame(columns=columns, dtype=np.float64)
      df_meta = df_meta.astype({'c': np.int64, 'r': np.int64})
      df_meta.index.name = 'feature'
  
      return dd.from_delayed(dfs, meta=df_meta)
  
  def extract_features(chunk, offsets, depth_xy):
      r, c, z, y, x = np.nonzero(chunk) 
      df = pd.DataFrame({'r': r, 'c': c, 'z': z, 'y': y+offsets[3]-depth_xy, 
                         'x': x+offsets[4]-depth_xy})
      df = df[(df.y &gt; depth_xy) &amp; (df.y &lt; (chunk.shape[3] - depth_xy)) &amp;
              (df.z &gt; depth_xy) &amp; (df.z &lt; (chunk.shape[4] - depth_xy))]
      return df
  
  data = np.zeros((2,4,8,16,16))  # round, channel, z, y, x
  data[0,0,4,2,2] = 1
  data[0,1,4,6,2] = 1
  data[1,2,4,8,2] = 1
  data[0,3,4,2,2] = 1
  arr = da.from_array(data)
  df = my_overlap(arr, chunk_xy=8, depth_xy=2)
  df.compute().reset_index()
  </code></pre>