title: Error while exporting a dask dataframe to csv
body: <p>My dask dataframe has about 120 mm rows and 4 columns:</p>
  
  <pre><code>df_final.dtypes
  
  cust_id        int64
  score            float64
  total_qty        float64
  update_score    float64
  dtype: object
  </code></pre>
  
  <p>and I'm doing this operation on jupyter notebooks connected to linux machine :</p>
  
  <pre><code>%time df_final.to_csv('/path/claritin-files-*.csv')
  </code></pre>
  
  <p>and it throws up this error:</p>
  
  <pre><code>---------------------------------------------------------------------------
  ValueError                                Traceback (most recent call last)
  &lt;ipython-input-24-46468ae45023&gt; in &lt;module&gt;()
  ----&gt; 1 get_ipython().magic(u"time df_final.to_csv('path/claritin-files-*.csv')")
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/IPython/core/interactiveshell.pyc in magic(self, arg_s)
     2334         magic_name, _, magic_arg_s = arg_s.partition(' ')
     2335         magic_name = magic_name.lstrip(prefilter.ESC_MAGIC)
  -&gt; 2336         return self.run_line_magic(magic_name, magic_arg_s)
     2337 
     2338     #-------------------------------------------------------------------------
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/IPython/core/interactiveshell.pyc in run_line_magic(self, magic_name, line)
     2255                 kwargs['local_ns'] = sys._getframe(stack_depth).f_locals
     2256             with self.builtin_trap:
  -&gt; 2257                 result = fn(*args,**kwargs)
     2258             return result
     2259 
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/IPython/core/magics/execution.pyc in time(self, line, cell, local_ns)
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/IPython/core/magic.pyc in &lt;lambda&gt;(f, *a, **k)
      191     **# but it's overkill for just that one bit of state.**
      192     def magic_deco(arg):
  --&gt; 193         call = lambda f, *a, **k: f(*a, **k)
      194 
      195         if callable(arg):
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/IPython/core/magics/execution.pyc in time(self, line, cell, local_ns)
     1161         if mode=='eval':
     1162             st = clock2()
  -&gt; 1163             out = eval(code, glob, local_ns)
     1164             end = clock2()
     1165         else:
  
  &lt;timed eval&gt; in &lt;module&gt;()
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/dask/dataframe/core.pyc in to_csv(self, filename, **kwargs)
      936         """ See dd.to_csv docstring for more information """
      937         from .io import to_csv
  --&gt; 938         return to_csv(self, filename, **kwargs)
      939 
      940     def to_delayed(self):
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/dask/dataframe/io/csv.pyc in to_csv(df, filename, name_function, compression, compute, get, **kwargs)
      411     if compute:
      412         from dask import compute
  --&gt; 413         compute(*values, get=get)
      414     else:
      415         return values
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/dask/base.pyc in compute(*args, **kwargs)
      177         dsk = merge(var.dask for var in variables)
      178     keys = [var._keys() for var in variables]
  --&gt; 179     results = get(dsk, keys, **kwargs)
      180 
      181     results_iter = iter(results)
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/dask/threaded.pyc in get(dsk, result, cache, num_workers, **kwargs)
       74     results = get_async(pool.apply_async, len(pool._pool), dsk, result,
       75                         cache=cache, get_id=_thread_get_id,
  ---&gt; 76                         **kwargs)
       77 
       78     # Cleanup pools associated to dead threads
  
  /home/mspra/anaconda2/lib/python2.7/site-packages/dask/async.pyc in get_async(apply_async, num_workers, dsk, result, cache, get_id, raise_on_exception, rerun_exceptions_locally, callbacks, dumps, loads, **kwargs)
      491                     _execute_task(task, data)  # Re-execute locally
      492                 else:
  --&gt; 493                     raise(remote_exception(res, tb))
      494             state['cache'][key] = res
      495             finish_task(dsk, key, state, results, keyorder.get)
  
  **ValueError: invalid literal for long() with base 10: 'total_qty'**
  
  Traceback
  ---------
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/dask/async.py", line 268, in execute_task
      result = _execute_task(task, data)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/dask/async.py", line 249, in _execute_task
      return func(*args2)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/dask/dataframe/io/csv.py", line 55, in pandas_read_text
      coerce_dtypes(df, dtypes)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/dask/dataframe/io/csv.py", line 83, in coerce_dtypes
      df[c] = df[c].astype(dtypes[c])
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/pandas/core/generic.py", line 3054, in astype
      raise_on_error=raise_on_error, **kwargs)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 3189, in astype
      return self.apply('astype', dtype=dtype, **kwargs)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 3056, in apply
      applied = getattr(b, f)(**kwargs)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 461, in astype
      values=values, **kwargs)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/pandas/core/internals.py", line 504, in _astype
      values = _astype_nansafe(values.ravel(), dtype, copy=True)
    File "/home/mspra/anaconda2/lib/python2.7/site-packages/pandas/types/cast.py", line 534, in _astype_nansafe
      return lib.astype_intsafe(arr.ravel(), dtype).reshape(arr.shape)
    File "pandas/lib.pyx", line 980, in pandas.lib.astype_intsafe (pandas/lib.c:17409)
    File "pandas/src/util.pxd", line 93, in util.set_value_at_unsafe (pandas/lib.c:72777)
  </code></pre>
  
  <p>I have a couple of questions:</p>
  
  <p>1) First of all this export was working fine on Friday, it spit out 100 csv files ( since it has 100 partitions),  which I later aggregated. So what is wrong today -- anything from the error log? </p>
  
  <p>2) May be this question is for the creators of this package, what is the most time-efficient way to get a csv extract out of a dask dataframe of this size, since it was taking about 1.5 to 2 hrs, the last time it was working. </p>
  
  <p>I'm not using dask distributed and this is on single core of a linux cluster. </p>