title: Dask bag from multiple files into Dask dataframe with columns
body: <p>I am given a list of filenames <code>files</code> which contain comma-delimited data which has to be cleaned as well as further extended by columns containing information based on the filenames. Thus, I implemented a small <code>read_file</code> function, which handles both, the initial cleaning, as well as the computation of additional columns. Using <code>db.from_sequence(files).map(read_file)</code>, I am mapping the read function to all of the files, getting a list of dictionaries each. </p>
  
  <p>However, rather than a list of dictionaries, I want my bag to contain each individual line of the input files as an entry. Subsequently, I want to map the keys of the dictionaries to column names in a dask dataframe.</p>
  
  <pre><code>from dask import bag as db
  
  def read_file(filename):
      ret = []
      with open(filename, 'r') as fp:
          ... # reading line of file and storing result in dict
          ret.append({'a': val_a, 'b': val_b, 'c': val_c})
  
      return ret
  
  from dask import bag as db
  files = ['a.txt', 'b.txt', 'c.txt']
  my_bag = db.from_sequence(files).map(read_file)
  # a,b,c are the keys of the dictionaries returned by read_file
  my_df = my_bag.to_dataframe(columns=['a', 'b', 'c'])
  </code></pre>
  
  <p>Could someone let me know what I have to change to get this code running? Are there different approaches that would be more suitable?</p>
  
  <p>Edit:
  I have created three test files <code>a_20160101.txt</code>, <code>a_20160102.txt</code>, <code>a_20160103.txt</code>. All of them contain just a few lines with a single string each.</p>
  
  <pre><code>asdf
  sadfsadf
  sadf
  fsadff
  asdf
  sadfasd
  fa
  sf
  ads 
  f
  </code></pre>
  
  <p>Previously I had a small error in <code>read_file</code>, but now, calling <code>my_bag.take(10)</code> after mapping to the reader works fine:</p>
  
  <pre><code>([{'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'asdf', 'c': 'XY'}, {'b':    datetime.datetime(2016, 2, 1, 0, 0), 'a': 'sadfsadf', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'sadf', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'fsadff', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'asdf', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'sadfasd', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'fa', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'sf', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'ads', 'c': 'XY'}, {'b': datetime.datetime(2016, 2, 1, 0, 0), 'a': 'f', 'c': 'XY'}],)
  </code></pre>
  
  <p>However <code>my_df = my_bag.to_dataframe(columns=['a', 'b', 'c'])</code> and subsequently
  <code>my_df.head(10)</code> still raises <code>dask.async.AssertionError: 3 columns passed, passed data had 10 columns</code></p>