title: 'Series' object has no attribute 'columns' in Dask
body: <p>I have a function <code>para_func</code> that takes a dataframe as input and returns a dataframe. I would like to group rows in <code>df</code> and apply function <code>para_func</code>. Finally, I have an error</p>
  <pre><code>distributed.worker - WARNING -  Compute Failed
  Function:  execute_task
  args:      ((subgraph_callable, 'root', None, &lt;function para_func at 0x000002F044CAC1F0&gt;, 'drop_by_shallow_copy-b7fa028b7c197981e52eea9f870f36c8', (&lt;function _concat at 0x000002F030567430&gt;, [            id parent_id     root  _partitions
  0      cqug90j   cqug2sr  cqug2sr            0
  1      cqug90k    34fvry   34fvry            0
  2      cqug90z   cqu80zb  cqu80zb            0
  3      cqug91c   cqtdj4m  cqtdj4m            0
  4      cqug91e   cquc4rc  cquc4rc            0
  ...        ...       ...      ...          ...
  99995  cqv8wz8   cqv1gg9   34hylq            0
  99996  cqv8wzj    34i1r5   34i1r5            0
  99997  cqv8wzv    34jasa   34jasa            0
  99998  cqv8wzx   cqv8k2k   34jasa            0
  99999  cqv8x08   cquywos   34hywb            0
  
  [100000 rows x 4 columns]], False), ['_partitions'], 'simple-shuffle-ead8404542740024e1572ac449733a42'))
  kwargs:    {}
  Exception: AttributeError(&quot;'Series' object has no attribute 'columns'&quot;)
  </code></pre>
  <p>Could you please elaborate on how to solve the error?</p>
  <pre><code>import pandas as pd
  import networkx as nx
  from dask.distributed import Client
  import dask.dataframe as dd
  client = Client(n_workers=4, threads_per_worker=2, processes=False, memory_limit='20GB')
  
  def para_func(tmp_df):
  
      siblings = pd.DataFrame({'id': tmp_df['id'], 'num_siblings': tmp_df.groupby('parent_id')['parent_id'].transform('count') - 1})
  
      children = tmp_df.groupby(by = 'id').size().reindex(tmp_df['id'], fill_value = 0).to_frame().reset_index(level = 0).rename(columns = {0: 'num_children'})
  
      att_df = siblings.merge(children, how = 'left', on = 'id')
      
      return att_df
  
  path = r'https://raw.githubusercontent.com/leanhdung1994/WebMining/main/sample_df2.csv'
  df = dd.read_csv(path, header = 0)
  
  result = df.groupby('root').apply(para_func, meta = object)
  computed_result = result.compute()
  </code></pre>