Repository: dask/dask

Issue:
Can't convert bag -> dataframe -> bag after flatten()
When I try to use `flatten()/concat()` on a bag and then proceed to convert to dataframe and back, the `pd.DataFrame` constructor fails because the data is passed in as an Iterator. See example below. Commenting out `.flatten()` or `.to_bag()` both cause it to be successful. But as written it causes an exception.

```
import dask.bag as db

(db
 .from_sequence(5 * [range(10)])
 .flatten()
 .to_dataframe()
 .to_bag()
 .compute()
)
```
```
TypeError                                 Traceback (most recent call last)
<ipython-input-1-31fb669b888e> in <module>()
      2 
      3 (db
----> 4  .from_sequence(5 * [range(10)])
      5  .flatten()
      6  .to_dataframe()

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/base.pyc in compute(self, **kwargs)
     95             Extra keywords to forward to the scheduler ``get`` function.
     96         """
---> 97         (result,) = compute(self, traverse=False, **kwargs)
     98         return result
     99 

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/base.pyc in compute(*args, **kwargs)
    202     dsk = collections_to_dsk(variables, optimize_graph, **kwargs)
    203     keys = [var._keys() for var in variables]
--> 204     results = get(dsk, keys, **kwargs)
    205 
    206     results_iter = iter(results)

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/multiprocessing.pyc in get(dsk, keys, num_workers, func_loads, func_dumps, optimize_graph, **kwargs)
    175                            get_id=_process_get_id, dumps=dumps, loads=loads,
    176                            pack_exception=pack_exception,
--> 177                            raise_exception=reraise, **kwargs)
    178     finally:
    179         if cleanup:

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in get_async(apply_async, num_workers, dsk, result, cache, get_id, rerun_exceptions_locally, pack_exception, raise_exception, callbacks, dumps, loads, **kwargs)
    519                         _execute_task(task, data)  # Re-execute locally
    520                     else:
--> 521                         raise_exception(exc, tb)
    522                 res, worker_id = loads(res_info)
    523                 state['cache'][key] = res

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in execute_task()
    288     try:
    289         task, data = loads(task_info)
--> 290         result = _execute_task(task, data)
    291         id = get_id()
    292         result = dumps((result, id))

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in _execute_task()
    268     elif istask(arg):
    269         func, args = arg[0], arg[1:]
--> 270         args2 = [_execute_task(a, cache) for a in args]
    271         return func(*args2)
    272     elif not ishashable(arg):

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/local.pyc in _execute_task()
    269         func, args = arg[0], arg[1:]
    270         args2 = [_execute_task(a, cache) for a in args]
--> 271         return func(*args2)
    272     elif not ishashable(arg):
    273         return arg

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/dask/bag/core.pyc in to_dataframe()
   1988 def to_dataframe(seq, columns, dtypes):
   1989     import pandas as pd
-> 1990     res = pd.DataFrame(seq, columns=list(columns))
   1991     return res.astype(dtypes, copy=False)

/Users/lsglick/.virtualenvs/test/lib/python2.7/site-packages/pandas/core/frame.pyc in __init__()
    323                 mgr = self._init_dict({}, index, columns, dtype=dtype)
    324         elif isinstance(data, collections.Iterator):
--> 325             raise TypeError("data argument can't be an iterator")
    326         else:
    327             try:

TypeError: data argument can't be an iterator
```

Versions:
Python: 3.5.2
Pandas: 0.20.2
Dask: 0.15.0

AND

Python: 2.7.10
Pandas: 0.19.2
Dask: 0.15.0

Discussion:

Patch:
diff --git a/dask/bag/core.py b/dask/bag/core.py
index aed4e6ca2..1cb43c01b 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1987,5 +1987,5 @@ def split(seq, n):
 
 def to_dataframe(seq, columns, dtypes):
     import pandas as pd
-    res = pd.DataFrame(seq, columns=list(columns))
+    res = pd.DataFrame(reify(seq), columns=list(columns))
     return res.astype(dtypes, copy=False)
diff --git a/dask/dataframe/io/io.py b/dask/dataframe/io/io.py
index 93b47bb3c..5c82683aa 100644
--- a/dask/dataframe/io/io.py
+++ b/dask/dataframe/io/io.py
@@ -343,7 +343,7 @@ def dataframe_from_ctable(x, slc, columns=None, categories=None, lock=lock):
 
 
 def from_dask_array(x, columns=None):
-    """ Create Dask Array from a Dask DataFrame
+    """ Create a Dask DataFrame from a Dask Array.
 
     Converts a 2d array into a DataFrame and a 1d array into a Series.
 
diff --git a/dask/delayed.py b/dask/delayed.py
index a9cd531e0..8f0943d0d 100644
--- a/dask/delayed.py
+++ b/dask/delayed.py
@@ -64,26 +64,33 @@ def to_task_dask(expr):
     """
     if isinstance(expr, Delayed):
         return expr.key, expr.dask
+
     if isinstance(expr, base.Base):
         name = 'finalize-' + tokenize(expr, pure=True)
         keys = expr._keys()
         dsk = expr._optimize(ensure_dict(expr.dask), keys)
         dsk[name] = (expr._finalize, (concrete, keys))
         return name, dsk
-    if isinstance(expr, tuple) and type(expr) != tuple:
-        return expr, {}
-    if isinstance(expr, (Iterator, list, tuple, set)):
+
+    if isinstance(expr, Iterator):
+        expr = list(expr)
+    typ = type(expr)
+
+    if typ in (list, tuple, set):
         args, dasks = unzip((to_task_dask(e) for e in expr), 2)
         args = list(args)
         dsk = sharedict.merge(*dasks)
         # Ensure output type matches input type
-        if isinstance(expr, (tuple, set)):
-            return (type(expr), args), dsk
-        else:
-            return args, dsk
-    if type(expr) is dict:
+        return (args, dsk) if typ is list else ((typ, args), dsk)
+
+    if typ is dict:
         args, dsk = to_task_dask([[k, v] for k, v in expr.items()])
         return (dict, args), dsk
+
+    if typ is slice:
+        args, dsk = to_task_dask([expr.start, expr.stop, expr.step])
+        return (slice,) + tuple(args), dsk
+
     return expr, {}
 
 


Test patch:
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index bd3679cbe..0053dc0b6 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -822,6 +822,13 @@ def test_to_dataframe():
     dd.utils.assert_eq(df, sol, check_index=False)
     check_parts(df, sol)
 
+    # Works with iterators
+    b = db.from_sequence(range(100), npartitions=5).map_partitions(iter)
+    sol = pd.DataFrame({'a': range(100)})
+    df = b.to_dataframe(columns=sol)
+    dd.utils.assert_eq(df, sol, check_index=False)
+    check_parts(df, sol)
+
 
 ext_open = [('gz', GzipFile), ('', open)]
 if not PY2:
diff --git a/dask/tests/test_delayed.py b/dask/tests/test_delayed.py
index 837e739a4..8c89cca23 100644
--- a/dask/tests/test_delayed.py
+++ b/dask/tests/test_delayed.py
@@ -1,4 +1,4 @@
-from collections import Iterator, namedtuple
+from collections import namedtuple
 from operator import add, setitem
 import pickle
 from random import random
@@ -32,6 +32,10 @@ def test_to_task_dask():
     assert task == x
     assert dict(dask) == {}
 
+    task, dask = to_task_dask(slice(a, b, 3))
+    assert task == (slice, 'a', 'b', 3)
+    assert dict(dask) == merge(a.dask, b.dask)
+
     # Issue https://github.com/dask/dask/issues/2107
     class MyClass(dict):
         pass
@@ -58,6 +62,8 @@ def test_operators():
     a = delayed([1, 2, 3])
     assert a[0].compute() == 1
     assert (a + a).compute() == [1, 2, 3, 1, 2, 3]
+    b = delayed(2)
+    assert a[:b].compute() == [1, 2]
 
     a = delayed(10)
     assert (a + 1).compute() == 11
@@ -154,7 +160,6 @@ def test_lists_are_concrete():
     assert c.compute() == 20
 
 
-@pytest.mark.xfail
 def test_iterators():
     a = delayed(1)
     b = delayed(2)
@@ -163,7 +168,6 @@ def test_iterators():
     assert c.compute() == 3
 
     def f(seq):
-        assert isinstance(seq, Iterator)
         return sum(seq)
 
     c = delayed(f)(iter([a, b]))