title: Dask apply with custom function
body: <p>I am experimenting with Dask, but I encountered a problem while using <code>apply</code> after grouping.</p>
  
  <p>I have a Dask DataFrame with a large number of rows. Let's consider for example the following</p>
  
  <pre class="lang-py prettyprint-override"><code>N=10000
  df = pd.DataFrame({'col_1':np.random.random(N), 'col_2': np.random.random(N) })
  ddf = dd.from_pandas(df, npartitions=8)
  </code></pre>
  
  <p>I want to bin the values of <code>col_1</code> and I follow the solution from <a href="https://stackoverflow.com/questions/47125665/simple-dask-map-partitions-example">here</a></p>
  
  <pre class="lang-py prettyprint-override"><code>bins = np.linspace(0,1,11)
  labels = list(range(len(bins)-1))
  ddf2 = ddf.map_partitions(test_f, 'col_1',bins,labels)
  </code></pre>
  
  <p>where</p>
  
  <pre class="lang-py prettyprint-override"><code>def test_f(df,col,bins,labels):
      return df.assign(bin_num = pd.cut(df[col],bins,labels=labels))
  </code></pre>
  
  <p>and this works as I expect it to.</p>
  
  <p>Now I want to take the median value in each bin (taken from <a href="https://github.com/dask/dask/issues/4362#issuecomment-453179565" rel="nofollow noreferrer">here</a>)</p>
  
  <pre class="lang-py prettyprint-override"><code>median = ddf2.groupby('bin_num')['col_1'].apply(pd.Series.median).compute()
  </code></pre>
  
  <p>Having 10 bins, I expect <code>median</code> to have 10 rows, but it actually has 80. The dataframe has 8 partitions so I guess that somehow the apply is working on each one individually.</p>
  
  <p>However, If I want the mean and use <code>mean</code></p>
  
  <pre class="lang-py prettyprint-override"><code>median = ddf2.groupby('bin_num')['col_1'].mean().compute()
  </code></pre>
  
  <p>it works and the output has 10 rows.</p>
  
  <p>The question is then: <em>what am I doing wrong that is preventing <code>apply</code> from operating  as <code>mean</code>?</em></p>