Repository: dask/dask

Issue:
to_delayed always optimizes
The `da.Array.to_delayed` call optimizes the underlying dask array's task graph before creating the dask.delayed objects.  In some cases this is not desired.  One option here would be to have an `optimize_graph=True` keyword argument.  There might also be broader solutions?

Discussion:
jcrist: cc @mrocklin.
mrocklin: This looks fine to me.  The similarity between the methods makes me want to find a way to put it on the Base class, although obviously there are sufficient dissimilarities with types like Array and Scalar that this probably isn't feasible.
jcrist: > The similarity between the methods makes me want to find a way to put it on the Base class, although obviously there are sufficient dissimilarities with types like Array and Scalar that this probably isn't feasible.

Yeah, I don't think this is feasible. Simple enough to reimplement for each collection.

One generic thing I did think about adding is the following function:

```python
def to_delayed(*args, **kwargs):
    """Convert multiple dask collections to delayed objects at the same time

    Parameters
    ------------
    optimize_graph : bool, optional
        If True (default), optimize all collections together before converting
        to delayed objects.
    """
    if kwargs.pop('optimize_graph', True):
         args = dask.optimize(*args)
    return tuple(a.to_delayed(optimize_graph=False) for a in args)
```

I decided against it because:

- This relies on collections having the same `to_delayed` signature
- Different collections return different things (lists of partitions, arrays of partitions, etc...). Grouping these all together felt off.

It's small enough to implement that I didn't think it was worth adding until a need arises. Can always add it later, but it's harder to remove things.

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 254232621..edbf36f8a 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -1833,10 +1833,14 @@ class Array(Base):
         memo[id(self)] = c
         return c
 
-    def to_delayed(self):
-        """ Convert Array into dask Delayed objects
+    def to_delayed(self, optimize_graph=True):
+        """Convert into an array of ``dask.delayed`` objects, one per chunk.
 
-        Returns an array of values, one value per chunk.
+        Parameters
+        ----------
+        optimize_graph : bool, optional
+            If True [default], the graph is optimized before converting into
+            ``dask.delayed`` objects.
 
         See Also
         --------
@@ -1844,7 +1848,9 @@ class Array(Base):
         """
         from ..delayed import Delayed
         keys = self.__dask_keys__()
-        dsk = self.__dask_optimize__(self.__dask_graph__(), keys)
+        dsk = self.__dask_graph__()
+        if optimize_graph:
+            dsk = self.__dask_optimize__(dsk, keys)
         L = ndeepmap(self.ndim, lambda k: Delayed(k, dsk), keys)
         return np.array(L, dtype=object)
 
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 27a0c6a6e..5c002d1e0 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -339,14 +339,19 @@ class Item(Base):
 
     __int__ = __float__ = __complex__ = __bool__ = Base.compute
 
-    def to_delayed(self):
-        """ Convert bag item to dask.delayed.
+    def to_delayed(self, optimize_graph=True):
+        """Convert into a ``dask.delayed`` object.
 
-        Returns a single value.
+        Parameters
+        ----------
+        optimize_graph : bool, optional
+            If True [default], the graph is optimized before converting into
+            ``dask.delayed`` objects.
         """
         from dask.delayed import Delayed
-        dsk = self.__dask_optimize__(self.__dask_graph__(),
-                                     self.__dask_keys__())
+        dsk = self.__dask_graph__()
+        if optimize_graph:
+            dsk = self.__dask_optimize__(dsk, self.__dask_keys__())
         return Delayed(self.key, dsk)
 
 
@@ -1260,14 +1265,24 @@ class Bag(Base):
         divisions = [None] * (self.npartitions + 1)
         return dd.DataFrame(dsk, name, meta, divisions)
 
-    def to_delayed(self):
-        """ Convert bag to list of dask Delayed.
+    def to_delayed(self, optimize_graph=True):
+        """Convert into a list of ``dask.delayed`` objects, one per partition.
 
-        Returns list of Delayed, one per partition.
+        Parameters
+        ----------
+        optimize_graph : bool, optional
+            If True [default], the graph is optimized before converting into
+            ``dask.delayed`` objects.
+
+        See Also
+        --------
+        dask.bag.from_delayed
         """
         from dask.delayed import Delayed
         keys = self.__dask_keys__()
-        dsk = self.__dask_optimize__(self.__dask_graph__(), keys)
+        dsk = self.__dask_graph__()
+        if optimize_graph:
+            dsk = self.__dask_optimize__(dsk, keys)
         return [Delayed(k, dsk) for k in keys]
 
     def repartition(self, npartitions):
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index e857330a3..f8abd5a8d 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -20,7 +20,7 @@ from .. import array as da
 from .. import core
 from ..utils import partial_by_order
 from .. import threaded
-from ..compatibility import apply, operator_div, bind_method, PY3
+from ..compatibility import apply, operator_div, bind_method
 from ..context import globalmethod
 from ..utils import (random_state_data, pseudorandom, derived_from, funcname,
                      memory_repr, put_lines, M, key_split, OperatorMethodMixin)
@@ -179,6 +179,21 @@ class Scalar(Base, OperatorMethodMixin):
     def _get_binary_operator(cls, op, inv=False):
         return lambda self, other: _scalar_binary(op, self, other, inv=inv)
 
+    def to_delayed(self, optimize_graph=True):
+        """Convert into a ``dask.delayed`` object.
+
+        Parameters
+        ----------
+        optimize_graph : bool, optional
+            If True [default], the graph is optimized before converting into
+            ``dask.delayed`` objects.
+        """
+        from dask.delayed import Delayed
+        dsk = self.__dask_graph__()
+        if optimize_graph:
+            dsk = self.__dask_optimize__(dsk, self.__dask_keys__())
+        return Delayed(self.key, dsk)
+
 
 def _scalar_binary(op, self, other, inv=False):
     name = '{0}-{1}'.format(funcname(op), tokenize(self, other))
@@ -1014,9 +1029,29 @@ Dask Name: {name}, {task} tasks""".format(klass=self.__class__.__name__,
         from .io import to_csv
         return to_csv(self, filename, **kwargs)
 
-    def to_delayed(self):
-        """ See dd.to_delayed docstring for more information """
-        return to_delayed(self)
+    def to_delayed(self, optimize_graph=True):
+        """Convert into a list of ``dask.delayed`` objects, one per partition.
+
+        Parameters
+        ----------
+        optimize_graph : bool, optional
+            If True [default], the graph is optimized before converting into
+            ``dask.delayed`` objects.
+
+        Examples
+        --------
+        >>> partitions = df.to_delayed()  # doctest: +SKIP
+
+        See Also
+        --------
+        dask.dataframe.from_delayed
+        """
+        from dask.delayed import Delayed
+        keys = self.__dask_keys__()
+        dsk = self.__dask_graph__()
+        if optimize_graph:
+            dsk = self.__dask_optimize__(dsk, keys)
+        return [Delayed(k, dsk) for k in keys]
 
     @classmethod
     def _get_unary_operator(cls, op):
@@ -4032,19 +4067,24 @@ def maybe_shift_divisions(df, periods, freq):
     return df
 
 
-def to_delayed(df):
-    """ Create Dask Delayed objects from a Dask Dataframe
+def to_delayed(df, optimize_graph=True):
+    """Convert into a list of ``dask.delayed`` objects, one per partition.
 
-    Returns a list of delayed values, one value per partition.
+    Deprecated, please use the equivalent ``df.to_delayed`` method instead.
 
-    Examples
+    Parameters
+    ----------
+    optimize_graph : bool, optional
+        If True [default], the graph is optimized before converting into
+        ``dask.delayed`` objects.
+
+    See Also
     --------
-    >>> partitions = df.to_delayed()  # doctest: +SKIP
+    dask.dataframe.from_delayed
     """
-    from dask.delayed import Delayed
-    keys = df.__dask_keys__()
-    dsk = df.__dask_optimize__(df.__dask_graph__(), keys)
-    return [Delayed(k, dsk) for k in keys]
+    warnings.warn("DeprecationWarning: The `dd.to_delayed` function is "
+                  "deprecated, please use the `.to_delayed()` method instead.")
+    return df.to_delayed(optimize_graph=optimize_graph)
 
 
 @wraps(pd.to_datetime)
@@ -4071,7 +4111,3 @@ def _repr_data_series(s, index):
     else:
         dtype = str(s.dtype)
     return pd.Series([dtype] + ['...'] * npartitions, index=index, name=s.name)
-
-
-if PY3:
-    _Frame.to_delayed.__doc__ = to_delayed.__doc__
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index dbdade336..d8759b91d 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -21,6 +21,8 @@ DataFrame
 - Avoid mutation in dataframe groupby tests (:pr:`3118`) `Matthew Rocklin`_
 - ``read_csv``, ``read_table``, and ``read_parquet`` accept iterables of paths
   (:pr:`3124`) `Jim Crist`_
+- Deprecates the ``dd.to_delayed`` *function* in favor of the existing method
+  (:pr:`3126`) `Jim Crist`_
 
 Bag
 +++
@@ -39,6 +41,8 @@ Core
   computing. (:pr:`3071`) `Jim Crist`_
 - Rename ``dask.optimize`` module to ``dask.optimization`` (:pr:`3071`) `Jim Crist`_
 - Change task ordering to do a full traversal (:pr:`3066`) `Matthew Rocklin`_
+- Adds an ``optimize_graph`` keyword to all ``to_delayed`` methods to allow
+  controlling whether optimizations occur on conversion. (:pr:`3126`) `Jim Crist`_
 
 
 0.16.1 / 2018-01-09
diff --git a/docs/source/dataframe-api.rst b/docs/source/dataframe-api.rst
index 87ee05fec..f50d99619 100644
--- a/docs/source/dataframe-api.rst
+++ b/docs/source/dataframe-api.rst
@@ -292,7 +292,6 @@ Store DataFrames
     to_hdf
     to_records
     to_bag
-    to_delayed
 
 DataFrame Methods
 ~~~~~~~~~~~~~~~~~
@@ -341,7 +340,6 @@ Storage and Conversion
 .. autofunction:: from_bcolz
 .. autofunction:: from_dask_array
 .. autofunction:: from_delayed
-.. autofunction:: to_delayed
 .. autofunction:: to_records
 .. autofunction:: to_csv
 .. autofunction:: to_bag
diff --git a/docs/source/dataframe-create.rst b/docs/source/dataframe-create.rst
index 7d429c3a2..759e4b0e6 100644
--- a/docs/source/dataframe-create.rst
+++ b/docs/source/dataframe-create.rst
@@ -39,7 +39,7 @@ Dask Collections:
     from_delayed
     from_dask_array
     dask.bag.core.Bag.to_dataframe
-    to_delayed
+    DataFrame.to_delayed
     to_records
     to_bag
 


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index a698df0bf..7d9e9492c 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -2370,12 +2370,20 @@ def test_to_delayed():
     assert a.compute() == s
 
 
-def test_to_delayed_optimizes():
+def test_to_delayed_optimize_graph():
     x = da.ones((4, 4), chunks=(2, 2))
     y = x[1:][1:][1:][:, 1:][:, 1:][:, 1:]
+
+    # optimizations
     d = y.to_delayed().flatten().tolist()[0]
     assert len([k for k in d.dask if k[0].startswith('getitem')]) == 1
 
+    # no optimizations
+    d2 = y.to_delayed(optimize_graph=False).flatten().tolist()[0]
+    assert dict(d2.dask) == dict(y.dask)
+
+    assert (d.compute() == d2.compute()).all()
+
 
 def test_cumulative():
     x = da.arange(20, chunks=5)
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index d3973630e..004139fba 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -18,6 +18,7 @@ from dask.bag.core import (Bag, lazify, lazify_task, map, collect,
                            reduceby, reify, partition, inline_singleton_lists,
                            optimize, from_delayed)
 from dask.compatibility import BZ2File, GzipFile, PY2
+from dask.delayed import Delayed
 from dask.utils import filetexts, tmpfile, tmpdir
 from dask.utils_test import inc, add
 
@@ -972,8 +973,6 @@ def test_bag_compute_forward_kwargs():
 
 
 def test_to_delayed():
-    from dask.delayed import Delayed
-
     b = db.from_sequence([1, 2, 3, 4, 5, 6], npartitions=3)
     a, b, c = b.map(inc).to_delayed()
     assert all(isinstance(x, Delayed) for x in [a, b, c])
@@ -985,17 +984,24 @@ def test_to_delayed():
     assert t.compute() == 21
 
 
-def test_to_delayed_optimizes():
+def test_to_delayed_optimize_graph():
     b = db.from_sequence([1, 2, 3, 4, 5, 6], npartitions=1)
     b2 = b.map(inc).map(inc).map(inc)
 
     [d] = b2.to_delayed()
     text = str(dict(d.dask))
     assert text.count('reify') == 1
+    [d2] = b2.to_delayed(optimize_graph=False)
+    assert dict(d2.dask) == dict(b2.dask)
+    assert d.compute() == d2.compute()
 
-    d = b2.sum().to_delayed()
+    x = b2.sum()
+    d = x.to_delayed()
     text = str(dict(d.dask))
     assert text.count('reify') == 0
+    d2 = x.to_delayed(optimize_graph=False)
+    assert dict(d2.dask) == dict(x.dask)
+    assert d.compute() == d2.compute()
 
     [d] = b2.to_textfiles('foo.txt', compute=False)
     text = str(dict(d.dask))
diff --git a/dask/dataframe/io/tests/test_io.py b/dask/dataframe/io/tests/test_io.py
index 45b034cd2..af6f1506b 100644
--- a/dask/dataframe/io/tests/test_io.py
+++ b/dask/dataframe/io/tests/test_io.py
@@ -528,17 +528,36 @@ def test_from_delayed_sorted():
 def test_to_delayed():
     df = pd.DataFrame({'x': [1, 2, 3, 4], 'y': [10, 20, 30, 40]})
     ddf = dd.from_pandas(df, npartitions=2)
+
+    # Frame
     a, b = ddf.to_delayed()
     assert isinstance(a, Delayed)
     assert isinstance(b, Delayed)
-
     assert_eq(a.compute(), df.iloc[:2])
 
+    # Scalar
+    x = ddf.x.sum()
+    dx = x.to_delayed()
+    assert isinstance(dx, Delayed)
+    assert_eq(dx.compute(), x)
+
 
-def test_to_delayed_optimizes():
+def test_to_delayed_optimize_graph():
     df = pd.DataFrame({'x': list(range(20))})
     ddf = dd.from_pandas(df, npartitions=20)
-    x = (ddf + 1).loc[:2]
+    ddf2 = (ddf + 1).loc[:2]
 
-    d = x.to_delayed()[0]
+    # Frame
+    d = ddf2.to_delayed()[0]
     assert len(d.dask) < 20
+    d2 = ddf2.to_delayed(optimize_graph=False)[0]
+    assert sorted(d2.dask) == sorted(ddf2.dask)
+    assert_eq(ddf2.get_partition(0), d.compute())
+    assert_eq(ddf2.get_partition(0), d2.compute())
+
+    # Scalar
+    x = ddf2.x.sum()
+    dx = x.to_delayed()
+    dx2 = x.to_delayed(optimize_graph=False)
+    assert len(dx.dask) < len(dx2.dask)
+    assert_eq(dx.compute(), dx2.compute())