Better error message on metadata inference failure (dask#1598)

Previously we'd just throw an error stating that metadata inference
failed. This made debugging the cause of these failures tricky. Now we
raise an error indicating that metadata inference failed, and include
the original error and traceback as well.

Author: jcrist

dask/dataframe/core.py

@@ -31,7 +31,8 @@
 from . import methods
 from .indexing import (_partition_of_index_value, _loc, _try_loc,
                        _coerce_loc_index, _maybe_partial_time_string)
-from .utils import meta_nonempty, make_meta, insert_meta_param_description
+from .utils import (meta_nonempty, make_meta, insert_meta_param_description,
+                    raise_on_meta_error)
 
 no_default = '__no_default__'
 
@@ -157,11 +158,7 @@ def _get_unary_operator(cls, op):
         def f(self):
             name = funcname(op) + '-' + tokenize(self)
             dsk = {(name, 0): (op, (self._name, 0))}
-            try:
-                meta = op(self._meta_nonempty)
-            except:
-                raise ValueError("Metadata inference failed in operator "
-                                 "{0}.".format(funcname(op)))
+            meta = op(self._meta_nonempty)
             return Scalar(merge(dsk, self.dask), name, meta)
         return f
 
@@ -189,16 +186,12 @@ def _scalar_binary(op, self, other, inv=False):
     else:
         dsk.update({(name, 0): (op, (self._name, 0), other_key)})
 
-    try:
-        other_meta = make_meta(other)
-        other_meta_nonempty = meta_nonempty(other_meta)
-        if inv:
-            meta = op(other_meta_nonempty, self._meta_nonempty)
-        else:
-            meta = op(self._meta_nonempty, other_meta_nonempty)
-    except:
-        raise ValueError("Metadata inference failed in operator "
-                         "{0}.".format(funcname(op)))
+    other_meta = make_meta(other)
+    other_meta_nonempty = meta_nonempty(other_meta)
+    if inv:
+        meta = op(other_meta_nonempty, self._meta_nonempty)
+    else:
+        meta = op(self._meta_nonempty, other_meta_nonempty)
 
     if return_type is not Scalar:
         return return_type(dsk, name, meta,
@@ -1631,11 +1624,7 @@ def map(self, arg, na_action=None, meta=no_default):
                    enumerate(self._keys()))
         dsk.update(self.dask)
         if meta is no_default:
-            try:
-                meta = self._meta_nonempty.map(arg, na_action=na_action)
-            except Exception:
-                raise ValueError("Metadata inference failed, please provide "
-                                 "`meta` keyword")
+            meta = _emulate(M.map, self, arg, na_action=na_action)
         else:
             meta = make_meta(meta)
 
@@ -1761,13 +1750,9 @@ def apply(self, func, convert_dtype=True, meta=no_default,
                    "  or:     .apply(func, meta=('x', 'f8'))            for series result")
             warnings.warn(msg)
 
-            try:
-                meta = _emulate(M.apply, self._meta_nonempty, func,
-                                convert_dtype=convert_dtype,
-                                args=args, **kwds)
-            except Exception:
-                raise ValueError("Metadata inference failed, please provide "
-                                 "`meta` keyword")
+            meta = _emulate(M.apply, self._meta_nonempty, func,
+                            convert_dtype=convert_dtype,
+                            args=args, **kwds)
 
         return map_partitions(M.apply, self, func,
                               convert_dtype, args, meta=meta, **kwds)
@@ -2350,12 +2335,8 @@ def apply(self, func, axis=0, args=(), meta=no_default,
                    "  or:     .apply(func, meta=('x', 'f8'))            for series result")
             warnings.warn(msg)
 
-            try:
-                meta = _emulate(M.apply, self._meta_nonempty, func,
-                                axis=axis, args=args, **kwds)
-            except Exception:
-                raise ValueError("Metadata inference failed, please provide "
-                                 "`meta` keyword")
+            meta = _emulate(M.apply, self._meta_nonempty, func,
+                            axis=axis, args=args, **kwds)
 
         return map_partitions(M.apply, self, func, axis,
                               False, False, None, args, meta=meta, **kwds)
@@ -2605,13 +2586,9 @@ def apply_concat_apply(args, chunk=None, aggregate=None, meta=no_default,
         dsk2 = {(b, 0): (apply, aggregate, [conc], aggregate_kwargs)}
 
     if meta is no_default:
-        try:
-            meta_chunk = _emulate(apply, chunk, args, chunk_kwargs)
-            meta = _emulate(apply, aggregate, [_concat([meta_chunk])],
-                            aggregate_kwargs)
-        except Exception:
-            raise ValueError("Metadata inference failed, please provide "
-                             "`meta` keyword")
+        meta_chunk = _emulate(apply, chunk, args, chunk_kwargs)
+        meta = _emulate(apply, aggregate, [_concat([meta_chunk])],
+                        aggregate_kwargs)
     meta = make_meta(meta)
 
     dasks = [arg.dask for arg in args if isinstance(arg, _Frame)]
@@ -2645,7 +2622,8 @@ def _emulate(func, *args, **kwargs):
     Apply a function using args / kwargs. If arguments contain dd.DataFrame /
     dd.Series, using internal cache (``_meta``) for calculation
     """
-    return func(*_extract_meta(args, True), **_extract_meta(kwargs, True))
+    with raise_on_meta_error(funcname(func)):
+        return func(*_extract_meta(args, True), **_extract_meta(kwargs, True))
 
 
 @insert_meta_param_description
@@ -2678,11 +2656,7 @@ def map_partitions(func, *args, **kwargs):
     args = _maybe_align_partitions(args)
 
     if meta is no_default:
-        try:
-            meta = _emulate(func, *args, **kwargs)
-        except Exception:
-            raise ValueError("Metadata inference failed, please provide "
-                             "`meta` keyword")
+        meta = _emulate(func, *args, **kwargs)
 
     if all(isinstance(arg, Scalar) for arg in args):
         dask = {(name, 0):

dask/dataframe/groupby.py

@@ -7,8 +7,8 @@
 
 from .core import DataFrame, Series, Index, aca, map_partitions, no_default
 from .shuffle import shuffle
-from .utils import make_meta, insert_meta_param_description
-from ..utils import derived_from, M
+from .utils import make_meta, insert_meta_param_description, raise_on_meta_error
+from ..utils import derived_from, M, funcname
 
 
 def _maybe_slice(grouped, columns):
@@ -324,11 +324,8 @@ def apply(self, func, meta=no_default, columns=no_default):
                    "  or:     .apply(func, meta=('x', 'f8'))            for series result")
             warnings.warn(msg)
 
-            try:
+            with raise_on_meta_error("groupby.apply({0})".format(funcname(func))):
                 meta = self._meta_nonempty.apply(func)
-            except:
-                raise ValueError("Metadata inference failed, please provide "
-                                 "`meta` keyword")
         else:
             meta = make_meta(meta)
 

dask/dataframe/tests/test_utils_dataframe.py

@@ -1,7 +1,8 @@
 import numpy as np
 import pandas as pd
 import dask.dataframe as dd
-from dask.dataframe.utils import shard_df_on_index, meta_nonempty, make_meta
+from dask.dataframe.utils import (shard_df_on_index, meta_nonempty, make_meta,
+                                  raise_on_meta_error)
 
 import pytest
 
@@ -187,3 +188,19 @@ def test_meta_nonempty_scalar():
     x = pd.Timestamp(2000, 1, 1)
     meta = meta_nonempty(x)
     assert meta is x
+
+
+def test_raise_on_meta_error():
+    try:
+        with raise_on_meta_error():
+            raise RuntimeError("Bad stuff")
+    except Exception as e:
+        assert e.args[0].startswith("Metadata inference failed.\n")
+        assert 'RuntimeError' in e.args[0]
+
+    try:
+        with raise_on_meta_error("myfunc"):
+            raise RuntimeError("Bad stuff")
+    except Exception as e:
+        assert e.args[0].startswith("Metadata inference failed in `myfunc`.\n")
+        assert 'RuntimeError' in e.args[0]

dask/dataframe/utils.py

@@ -5,6 +5,8 @@
 
 from collections import Iterator
 import sys
+import traceback
+from contextlib import contextmanager
 
 import numpy as np
 import pandas as pd
@@ -130,6 +132,33 @@ def insert_meta_param_description(*args, **kwargs):
     return f
 
 
+@contextmanager
+def raise_on_meta_error(funcname=None):
+    """Reraise errors in this block to show metadata inference failure.
+
+    Parameters
+    ----------
+    funcname : str, optional
+        If provided, will be added to the error message to indicate the
+        name of the method that failed.
+    """
+    try:
+        yield
+    except Exception as e:
+        exc_type, exc_value, exc_traceback = sys.exc_info()
+        tb = ''.join(traceback.format_tb(exc_traceback))
+        msg = ("Metadata inference failed{0}.\n\n"
+               "Original error is below:\n"
+               "------------------------\n"
+               "{1}\n\n"
+               "Traceback:\n"
+               "---------\n"
+               "{2}"
+               ).format(" in `{0}`".format(funcname) if funcname else "",
+                        repr(e), tb)
+        raise ValueError(msg)
+
+
 def make_meta(x, index=None):
     """Create an empty pandas object containing the desired metadata.