title: Dask Concatenate a Series of Dataframes
body: <p>I have a Dask Series of Pandas DataFrames. I would like to use <code>dask.dataframe.multi.concat</code> to convert this into a Dask DataFrame. However the <code>dask.dataframe.multi.concat</code> always requires a list of DataFrames.</p>
  
  <p>I could perform a <code>compute</code> on the Dask series of Pandas DataFrames to get a Pandas series of DataFrames, at which point I could turn that into a list. But I think it would be better not to call <code>compute</code> and instead directly acquire the Dask DataFrame from the Dask Series of Pandas DataFrames.</p>
  
  <p>What would the best way to do this? Here's my code that produces the series of dataframes</p>
  
  <pre><code>import pandas as pd
  import dask.dataframe as dd
  import operator
  import numpy as np
  import math
  import itertools
  
  def apportion_pcts(pcts, total):
      """Apportion an integer by percentages
      Uses the largest remainder method
      """
      if (sum(pcts) != 100):
          raise ValueError('Percentages must add up to 100')
      proportions = [total * (pct / 100) for pct in pcts]
      apportions = [math.floor(p) for p in proportions]
      remainder = total - sum(apportions)
      remainders = [(i, p - math.floor(p)) for (i, p) in enumerate(proportions)]
      remainders.sort(key=operator.itemgetter(1), reverse=True)
      for (i, _) in itertools.cycle(remainders):
          if remainder == 0:
              break
          else:
              apportions[i] += 1
              remainder -= 1
      return apportions
  
  
  # images_df = dd.read_csv('./tests/data/classification/images.csv')
  images_df = pd.DataFrame({"image_id": [0,1,2,3,4,5], "image_class_id": [0,1,1,3,3,5]})
  images_df = dd.from_pandas(images_df, npartitions=1)
  
  output_ratio = [80, 20]
  
  def partition_class (partition):
      size = len(partition)
      proportions = apportion_pcts(output_ratio, size)
      slices = []
      start = 0
      for proportion in proportions:
          s = slice(start, start + proportion)
          slices.append(partition.iloc[s, :])
          start = start+proportion
      slicess = pd.Series(slices)
      return slicess
  
  partitioned_schema = dd.utils.make_meta(
      [(0, object), (1, object)], pd.Index([], name='image_class_id'))
  partitioned_df = images_df.groupby('image_class_id')
  partitioned_df = partitioned_df.apply(partition_class, meta=partitioned_schema)
  </code></pre>
  
  <p>In the <code>partitioned_df</code>, we can get <code>partitioned_df[0]</code> or <code>partitioned_df[1]</code> to get a series of dataframe objects.</p>
  
  <hr>
  
  <p>Here is an example of the CSV file:</p>
  
  <pre><code>image_id,image_width,image_height,image_path,image_class_id
  0,224,224,tmp/data/image_matrices/0.npy,5
  1,224,224,tmp/data/image_matrices/1.npy,0
  2,224,224,tmp/data/image_matrices/2.npy,4
  3,224,224,tmp/data/image_matrices/3.npy,1
  4,224,224,tmp/data/image_matrices/4.npy,9
  5,224,224,tmp/data/image_matrices/5.npy,2
  6,224,224,tmp/data/image_matrices/6.npy,1
  7,224,224,tmp/data/image_matrices/7.npy,3
  8,224,224,tmp/data/image_matrices/8.npy,1
  9,224,224,tmp/data/image_matrices/9.npy,4
  </code></pre>
  
  <p>I tried to do a reduction afterwards, but this doesn't quite make sense due to a proxy <code>foo</code> string.</p>
  
  <pre><code>def zip_partitions(s):
      r = []
      for c in s.columns:
          l = s[c].tolist()
          r.append(pd.concat(l))
      return pd.Series(r)
  
  output_df = partitioned_df.reduction(
      chunk=zip_partitions
  )
  </code></pre>
  
  <p>The proxy list that I'm attempting to concat is <code>['foo', 'foo']</code>. What is this phase for? To discover how to do the task? But then certain operations don't work. I'm wondering if it's because I'm operating over objects that I'm getting these strings.</p>