title: convert dask.bag of dictionaries to dask.dataframe using dask.delayed and pandas.DataFrame
body: <p>I am struggling to convert a <code>dask.bag</code> of dictionaries into <code>dask.delayed</code> <code>pandas.DataFrames</code> into a final <code>dask.dataframe</code></p>
  
  <p>I have one function (make_dict) that reads files into a rather complex nested dictionary structure and another function (make_df) to turn these dictionaries into a <code>pandas.DataFrame</code> (resulting dataframe is around 100 mb for each file). I would like to append all dataframes into a single <code>dask.dataframe</code> for further analysis.</p>
  
  <p>Up to now I was using <code>dask.delayed</code> objects to load, convert and append all data which works fine (see example below). However for future work I would like to store the loaded dictionaries in a <code>dask.bag</code> using <code>dask.persist()</code>.</p>
  
  <p>I managed to load the data into <code>dask.bag</code>, resulting in a list of dicts or list of <code>pandas.DataFrame</code> that I can use locally after calling <code>compute()</code>. When I tried turning the <code>dask.bag</code> into a <code>dask.dataframe</code> using <code>to_delayed()</code> however, I got stuck with an error (see below).</p>
  
  <p>It feels like I am missing something rather simple here or maybe my approach to <code>dask.bag</code> is wrong?</p>
  
  <p>The below example shows my approach using simplified functions and throws the same error. Any advice on how to tackle this is appreciated.</p>
  
  <pre class="lang-py prettyprint-override"><code>import numpy as np
  import pandas as pd
  import dask
  import dask.dataframe
  import dask.bag
  
  print(dask.__version__) # 1.1.4
  print(pd.__version__) # 0.24.2
  
  def make_dict(n=1):
      return {"name":"dictionary","data":{'A':np.arange(n),'B':np.arange(n)}}
  
  def make_df(d):
      return pd.DataFrame(d['data'])
  
  k = [1,2,3]
  
  # using dask.delayed
  dfs = []
  for n in k:
      delayed_1 = dask.delayed(make_dict)(n)
      delayed_2 = dask.delayed(make_df)(delayed_1)
      dfs.append(delayed_2)
  ddf1 = dask.dataframe.from_delayed(dfs).compute() # this works as expected
  
  # using dask.bag and turning bag of dicts into bag of DataFrames
  b1 = dask.bag.from_sequence(k).map(make_dict)
  b2 = b1.map(make_df)
  
  df = pd.DataFrame().append(b2.compute()) # &lt;- I would like to do this using delayed dask.DataFrames like above
  ddf2 = dask.dataframe.from_delayed(b2.to_delayed()).compute() # &lt;- this fails
  
  # error:
  # ValueError: Expected iterable of tuples of (name, dtype), got [   A  B
  # 0  0  0]
  
  </code></pre>
  
  <p>what I ultimately would like to do using the distributed scheduler:</p>
  
  <pre class="lang-py prettyprint-override"><code>b = dask.bag.from_sequence(k).map(make_dict)
  b = b.persist()
  ddf = dask.dataframe.from_delayed(b.map(make_df).to_delayed())
  </code></pre>