title: Parallelization on cluster dask
body: <p>I'm looking for the best way to parallelize on a cluster the following problem. I have several files</p>
  
  <ul>
  <li>folder/file001.csv</li>
  <li>folder/file002.csv</li>
  <li>:</li>
  <li>folder/file100.csv</li>
  </ul>
  
  <p>They are disjoints with respect to the <code>key</code> I want to use to groupby, that is if a set of keys is in <code>file1.csv</code> any of these keys has an item in any other files.</p>
  
  <p>In one side I can just run </p>
  
  <pre><code>df = dd.read_csv("folder/*")
  df.groupby("key").apply(f, meta=meta).compute(scheduler='processes') 
  </code></pre>
  
  <p>But I'm wondering if there is a better/smarter way to do so in a sort of 
  <a href="https://github.com/TomAugspurger/dask-tutorial-pycon-2018/blob/master/solutions/01-delayed-groupby.py" rel="nofollow noreferrer">delayed-groupby</a> way.</p>
  
  <p>Every <code>filexxx.csv</code> fits in memory on a node. Given that every node has <code>n</code> cores it will be ideal use all of them. For every single file I can use this <a href="http://blog.adeel.io/2016/11/06/parallelize-pandas-map-or-apply/" rel="nofollow noreferrer">hacky</a> way</p>
  
  <pre><code>import numpy as np
  import multiprocessing as mp
  
  cores = mp.cpu_count() #Number of CPU cores on your system
  partitions = cores #Define as many partitions as you want
  
  def parallelize(data, func):
      data_split = np.array_split(data, partitions)
      pool = mp.Pool(cores)
      data = pd.concat(pool.map(func, data_split))
      pool.close()
      pool.join()
      return data
  
  data = parallelize(data, f);
  </code></pre>
  
  <p>And, again, I'm not sure if there is an efficent dask way to do so.</p>