title: Running Embarrasingly Parallel operations on a single piece of data using Dask
body: <p>I was following <a href="https://examples.dask.org/applications/embarrassingly-parallel.html" rel="nofollow noreferrer">this</a> tutorial and I was able to parallelize a for loop where operations were done independently on multiple files. However, now I need to perform an iterative function to extract variables into 300 files from a single xarray dataset (source). My functions looks like this :</p>
  <pre><code>def build_data(parameters)
      extract var1 from source
      extract var2 from source
      extract var3 from source
      ..........
      return('finished')
  </code></pre>
  <p>I have set 6 workers and to run this function in a loop 300 times, I implemented :</p>
  <pre><code>source = xr.opendataset()
  build_delayed = []
  for i in range (0,300):
     task = dask.delayed (build_data) (params)
     build_delayed.append(task)
  dask.compute(*build_delayed)
  </code></pre>
  <p>But this is not working and seems like latency is building up as all workers are trying access same piece of data. Also, when the loop ends, <code>None</code> is returned in build_delayed[ ].</p>
  <p>What am I missing here? How do I make it work in parallel in this case?</p>
  <p>(My detailed <a href="https://stackoverflow.com/questions/62971626/parallelizing-generic-python-code-with-dask">question</a>.)</p>
  <p>EDIT : MVCE :</p>
  <pre><code>stations = []
  for key in valid_stations:
      station = pd.read_csv('~/projects/LIS_verification/pyLISF/data/IFI-Observations/flow_2/'+key+'.csv')
      station.reset_index(drop = True,inplace=True)
      station.set_index('Date',inplace=True)
      station.index = pd.to_datetime(station.index)
      stations.append(station)
  
  routedat = xr.opendataset('/LDAS/02-ILDAS/OUTPUT/EXP002/SURFACEMODEL/file1.nc')
      
  def build_data(stations,routedat,i) :
      try :
          start_date = stations[i]['Streamflow (cumecs)'].first_valid_index()
          lis_date = routedat['time'].values
          return (start_date,lis_date)
      except Exception as e :
          return( start_date,str(e))
      
  for i in range (0,10):
      task = dask.delayed (build_data) (stations,routedat,i)
      build_delayed.append(task)
  
  dask.compute(*build_delayed)
  </code></pre>
  <p>I am getting the required output but the time taken is too large as compared to sequential loop (508 ms vs 31 seconds).</p>
  <p>UPDATE : I was able to run it successfully in &lt; 300 ms in parallel by using command <code>.compute(scheduler='synchronous')</code></p>