Repository: dask/dask

Issue:
Dask objects inspection failing, issues with graph
Hi,

- I cannot reproduce the example given [here](http://dask.pydata.org/en/latest/inspect.html) on the dask website in order to inspect dask objects.
Please see the following notebook:
[inspect dask object](https://github.com/apatlpo/lops-array/blob/master/sandbox/test_graph.ipynb)

(sorry if github is not rendering notebooks as it the case for me right now)

- Also when I try to print graphs, complicated numbers show up which complicate the interpretation. For example:
[graph](https://github.com/apatlpo/lops-array/blob/master/sandbox/natl60_tseries_debug.ipynb)

Any thoughts?
Are these relates issues?



Discussion:

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 9346b1c04..15a63bd8b 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2159,7 +2159,9 @@ def atop(func, out_ind, *args, **kwargs):
     concatenate : bool, keyword only
         If true concatenate arrays along dummy indices, else provide lists
     adjust_chunks : dict
-        Dictionary mapping index to function to be applied to chunk sizes
+        Dictionary mapping index to information to adjust chunk sizes.  Can
+        either be a constant chunksize, a tuple of all chunksizes, or a
+        function that converts old chunksize to new chunksize
     new_axes : dict, keyword only
         New indexes and their dimension lengths
 
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 4428e0ba5..62297b290 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1,16 +1,15 @@
 from __future__ import absolute_import, division, print_function
 
-from collections import Iterable, Iterator, defaultdict
-from functools import wraps, partial
 import itertools
 import math
-from operator import getitem
 import types
 import uuid
-from random import Random
-from warnings import warn
+import warnings
+from collections import Iterable, Iterator, defaultdict
 from distutils.version import LooseVersion
-
+from functools import wraps, partial
+from operator import getitem
+from random import Random
 
 from toolz import (merge, take, reduce, valmap, map, partition_all, filter,
                    remove, compose, curry, first, second, accumulate, peek)
@@ -1050,7 +1049,7 @@ class Bag(Base):
 
         return type(self)(merge(self.dask, dsk), e, 1)
 
-    def take(self, k, npartitions=1, compute=True):
+    def take(self, k, npartitions=1, compute=True, warn=True):
         """ Take the first k elements.
 
         Parameters
@@ -1064,6 +1063,9 @@ class Bag(Base):
             returned. Pass -1 to use all partitions.
         compute : bool, optional
             Whether to compute the result, default is True.
+        warn : bool, optional
+            Whether to warn if the number of elements returned is less than
+            requested, default is True.
 
         >>> b = from_sequence(range(10))
         >>> b.take(3)  # doctest: +SKIP
@@ -1087,9 +1089,9 @@ class Bag(Base):
                 dsk[(name_p, i)] = (list, (take, k, (self.name, i)))
 
             concat = (toolz.concat, ([(name_p, i) for i in range(npartitions)]))
-            dsk[(name, 0)] = (safe_take, k, concat)
+            dsk[(name, 0)] = (safe_take, k, concat, warn)
         else:
-            dsk = {(name, 0): (safe_take, k, (self.name, 0))}
+            dsk = {(name, 0): (safe_take, k, (self.name, 0), warn)}
 
         b = Bag(merge(self.dask, dsk), name, 1)
 
@@ -1215,12 +1217,16 @@ class Bag(Base):
         import dask.dataframe as dd
         if meta is None:
             if isinstance(columns, pd.DataFrame):
-                warn("Passing metadata to `columns` is deprecated. Please "
-                     "use the `meta` keyword instead.")
+                warnings.warn("Passing metadata to `columns` is deprecated. "
+                              "Please use the `meta` keyword instead.")
                 meta = columns
             else:
-                head = self.take(1)[0]
-                meta = pd.DataFrame([head], columns=columns)
+                head = self.take(1, warn=False)
+                if len(head) == 0:
+                    raise ValueError("`dask.bag.Bag.to_dataframe` failed to "
+                                     "properly infer metadata, please pass in "
+                                     "metadata via the `meta` keyword")
+                meta = pd.DataFrame(list(head), columns=columns)
         elif columns is not None:
             raise ValueError("Can't specify both `meta` and `columns`")
         else:
@@ -1986,12 +1992,12 @@ def empty_safe_aggregate(func, parts, is_last):
     return empty_safe_apply(func, parts2, is_last)
 
 
-def safe_take(n, b):
+def safe_take(n, b, warn=True):
     r = list(take(n, b))
-    if len(r) != n:
-        warn("Insufficient elements for `take`. {0} elements requested, "
-             "only {1} elements available. Try passing larger `npartitions` "
-             "to `take`.".format(n, len(r)))
+    if len(r) != n and warn:
+        warnings.warn("Insufficient elements for `take`. {0} elements "
+                      "requested, only {1} elements available. Try passing "
+                      "larger `npartitions` to `take`.".format(n, len(r)))
     return r
 
 
diff --git a/docs/source/inspect.rst b/docs/source/inspect.rst
index 2cd98f57a..4725af80a 100644
--- a/docs/source/inspect.rst
+++ b/docs/source/inspect.rst
@@ -19,7 +19,7 @@ The first step is to look at the ``.dask`` attribute of an array
 
    >>> import dask.array as da
    >>> x = da.ones((5, 15), chunks=(5, 5))
-   >>> x.dask
+   >>> dict(x.dask)
    {('wrapped_1', 0, 0): (ones, (5, 5)),
     ('wrapped_1', 0, 1): (ones, (5, 5)),
     ('wrapped_1', 0, 2): (ones, (5, 5))}
@@ -29,7 +29,7 @@ objects
 
 .. code-block:: python
 
-   >>> (x + 1).dask
+   >>> dict((x + 1).dask)
    {('wrapped_1', 0, 0): (ones, (5, 5)),
     ('wrapped_1', 0, 1): (ones, (5, 5)),
     ('wrapped_1', 0, 2): (ones, (5, 5))
@@ -37,6 +37,9 @@ objects
     ('x_1', 0, 1): (add, ('wrapped_1', 0, 1), 1),
     ('x_1', 0, 2): (add, ('wrapped_1', 0, 2), 1)}
 
+.. note:: In this example we use simple names like ``x_1``, ``ones``, and
+  ``add`` for demonstration purposes.  However in practice these names may be
+  more complex and include long hashed names.
 
 Visualize graphs with DOT
 -------------------------


Test patch:
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index 46427f714..44b874d48 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -530,17 +530,23 @@ def test_take_npartitions():
         b.take(1, npartitions=5)
 
 
-@pytest.mark.skipif(sys.version_info[:2] == (3,3),
-                    reason="Python3.3 uses pytest2.7.2, w/o warns method")
 def test_take_npartitions_warn():
-    with pytest.warns(None):
-        b.take(100)
+    # Use single-threaded scheduler so warnings are properly captured in the
+    # same process
+    with dask.set_options(get=dask.get):
+        with pytest.warns(UserWarning):
+            b.take(100)
+
+        with pytest.warns(UserWarning):
+            b.take(7)
 
-    with pytest.warns(None):
-        b.take(7)
+        with pytest.warns(None) as rec:
+            b.take(7, npartitions=2)
+        assert len(rec) == 0
 
-    with pytest.warns(None):
-        b.take(7, npartitions=2)
+        with pytest.warns(None) as rec:
+            b.take(7, warn=False)
+        assert len(rec) == 0
 
 
 def test_map_is_lazy():
@@ -775,6 +781,11 @@ def test_to_dataframe():
     with pytest.raises(ValueError):
         b.to_dataframe(columns=['a', 'b'], meta=sol)
 
+    # Inference fails if empty first partition
+    b2 = b.filter(lambda x: x['a'] > 200)
+    with pytest.raises(ValueError):
+        b2.to_dataframe()
+
     # Single column
     b = b.pluck('a')
     sol = sol[['a']]