Repository: dask/dask

Issue:
DataFrame: Splitting a list to multiple columns is throwing error 
I was trying to split a string and saved value in a temp column and use that column to populate additional columns.  I am able to split string and convert to a list. But while fetching elements from list and assigning to new column is throwing error. 

I have tried to recreate the error with below code:

```
df = pd.DataFrame({ 'a' : [1, 2, 3] ,
      'b' : ['a|b|c|d|e|f|g', 'a|b|c|d|e|f|g', 'a|b|c|d|e|f|g']
            })
ddf = dd.from_pandas(df,1)
ddf['c'] = ddf.b.str.split('|')
ddf = ddf.assign(col1= ddf.c.map(lambda x: x[0]))
ddf = ddf.assign(col2= ddf.c.map(lambda x: x[1]))
ddf = ddf.assign(col3= ddf.c.map(lambda x: x[2]))
ddf = ddf.assign(col4= ddf.c.map(lambda x: x[3]))
```
Getting error while executing last line >> ddf = ddf.assign(col4= ddf.c.map(lambda x: x[3]))
   IndexError('string index out of range',) 



Discussion:
jcrist: This error isn't only thrown for custom functions, this also shows up when the user does an operation that's invalid:

```pytb
In [1]: import dask.dataframe as dd

In [2]: import pandas as pd

In [3]: df = pd.DataFrame({'a': ['x', 'y', 'y'], 'b': ['x', 'y', 'z'], 'c': [1, 2, 3]})

In [4]: ddf = dd.from_pandas(df, npartitions=1)

In [5]: ddf.a + ddf.c
---------------------------------------------------------------------------
...
ValueError: Metadata inference failed in `add`.

Original error is below:
------------------------
TypeError("ufunc 'add' did not contain a loop with signature matching types dtype('<U21') dtype('<U21') dtype('<U21')",)

Traceback:
---------
  File "/Users/jcrist/anaconda/envs/dask/lib/python3.5/site-packages/dask-0.17.1-py3.5.egg/dask/dataframe/utils.py", line 137, in raise_on_meta_error
    yield
  File "/Users/jcrist/anaconda/envs/dask/lib/python3.5/site-packages/dask-0.17.1-py3.5.egg/dask/dataframe/core.py", line 3062, in elemwise
    meta = partial_by_order(*parts, function=op, other=other)
  File "/Users/jcrist/anaconda/envs/dask/lib/python3.5/site-packages/dask-0.17.1-py3.5.egg/dask/utils.py", line 879, in partial_by_order
    return function(*args2, **kwargs)
  File "/Users/jcrist/anaconda/envs/dask/lib/python3.5/site-packages/pandas/core/ops.py", line 739, in wrapper
    result = wrap_results(safe_na_op(lvalues, rvalues))
  File "/Users/jcrist/anaconda/envs/dask/lib/python3.5/site-packages/pandas/core/ops.py", line 710, in safe_na_op
    lambda x: op(x, rvalues))
  File "pandas/_libs/algos_common_helper.pxi", line 1212, in pandas._libs.algos.arrmap_object
  File "/Users/jcrist/anaconda/envs/dask/lib/python3.5/site-packages/pandas/core/ops.py", line 710, in <lambda>
    lambda x: op(x, rvalues))
```

I recommend adding a flag to `raise_on_meta_error` to optionally add this suggestion, and then set it appropriately in the proper places.

Patch:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index ff2e21e63..c4d6eb99f 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1970,7 +1970,7 @@ Dask Name: {name}, {task} tasks""".format(klass=self.__class__.__name__,
                enumerate(self.__dask_keys__())}
         dsk.update(self.dask)
         if meta is no_default:
-            meta = _emulate(M.map, self, arg, na_action=na_action)
+            meta = _emulate(M.map, self, arg, na_action=na_action, udf=True)
         else:
             meta = make_meta(meta)
 
@@ -2121,7 +2121,7 @@ Dask Name: {name}, {task} tasks""".format(klass=self.__class__.__name__,
 
             meta = _emulate(M.apply, self._meta_nonempty, func,
                             convert_dtype=convert_dtype,
-                            args=args, **kwds)
+                            args=args, udf=True, **kwds)
 
         return map_partitions(M.apply, self, func,
                               convert_dtype, args, meta=meta, **kwds)
@@ -2794,7 +2794,7 @@ class DataFrame(_Frame):
             warnings.warn(msg)
 
             meta = _emulate(M.apply, self._meta_nonempty, func,
-                            axis=axis, args=args, **kwds)
+                            axis=axis, args=args, udf=True, **kwds)
 
         return map_partitions(M.apply, self, func, axis,
                               False, False, None, args, meta=meta, **kwds)
@@ -3291,8 +3291,8 @@ def apply_concat_apply(args, chunk=None, aggregate=None, combine=None,
             dsk[(b, j)] = (aggregate, conc)
 
     if meta is no_default:
-        meta_chunk = _emulate(chunk, *args, **chunk_kwargs)
-        meta = _emulate(aggregate, _concat([meta_chunk]),
+        meta_chunk = _emulate(chunk, *args, udf=True, **chunk_kwargs)
+        meta = _emulate(aggregate, _concat([meta_chunk]), udf=True,
                         **aggregate_kwargs)
     meta = make_meta(meta)
 
@@ -3332,7 +3332,7 @@ def _emulate(func, *args, **kwargs):
     Apply a function using args / kwargs. If arguments contain dd.DataFrame /
     dd.Series, using internal cache (``_meta``) for calculation
     """
-    with raise_on_meta_error(funcname(func)):
+    with raise_on_meta_error(funcname(func), udf=kwargs.pop('udf', False)):
         return func(*_extract_meta(args, True), **_extract_meta(kwargs, True))
 
 
@@ -3369,7 +3369,7 @@ def map_partitions(func, *args, **kwargs):
     args = _maybe_align_partitions(args)
 
     if meta is no_default:
-        meta = _emulate(func, *args, **kwargs)
+        meta = _emulate(func, *args, udf=True, **kwargs)
 
     if all(isinstance(arg, Scalar) for arg in args):
         dask = {(name, 0):
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index 1f23badea..dbbc66fed 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -124,7 +124,7 @@ def insert_meta_param_description(*args, **kwargs):
 
 
 @contextmanager
-def raise_on_meta_error(funcname=None):
+def raise_on_meta_error(funcname=None, udf=False):
     """Reraise errors in this block to show metadata inference failure.
 
     Parameters
@@ -138,15 +138,19 @@ def raise_on_meta_error(funcname=None):
     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)
+        msg = "Metadata inference failed{0}.\n\n"
+        if udf:
+            msg += ("You have supplied a custom function and Dask is unable to \n"
+                    "determine the type of output that that function returns. \n\n"
+                    "To resolve this please provide a meta= keyword.\n"
+                    "The docstring of the Dask function you ran should have more information.\n\n")
+        msg += ("Original error is below:\n"
+                "------------------------\n"
+                "{1}\n\n"
+                "Traceback:\n"
+                "---------\n"
+                "{2}")
+        msg = msg.format(" in `{0}`".format(funcname) if funcname else "", repr(e), tb)
         raise ValueError(msg)
 
 
diff --git a/dask/delayed.py b/dask/delayed.py
index 484a45564..1da6e040c 100644
--- a/dask/delayed.py
+++ b/dask/delayed.py
@@ -132,7 +132,9 @@ def delayed(obj, name=None, pure=None, nout=None, traverse=True):
         The function or object to wrap
     name : string or hashable, optional
         The key to use in the underlying graph for the wrapped object. Defaults
-        to hashing content.
+        to hashing content. Note that this only affects the name of the object
+        wrapped by this call to delayed, and *not* the output of delayed
+        function calls - for that use ``dask_key_name=`` as described below.
     pure : bool, optional
         Indicates whether calling the resulting ``Delayed`` object is a pure
         operation. If True, arguments to the call are hashed to produce
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index fb39b4e1d..6d144c877 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -17,6 +17,7 @@ DataFrame
 - Allow `t` as shorthand for `table` in `to_hdf` for pandas compatibility `Jörg Dietrich`_
 - Added top level `isna` method for Dask DataFrames (:pr:`3294`) `Christopher Ren`_
 - Fix selection on partition column on ``read_parquet`` for ``engine="pyarrow"`` (:pr:`3207`) `Uwe Korn`_
+- Provide more informative error message for meta= errors (:pr:`3343`) `Matthew Rocklin`_
 
 Bag
 +++


Test patch:
diff --git a/dask/array/tests/test_routines.py b/dask/array/tests/test_routines.py
index 3e1c694b2..70bd0ea1c 100644
--- a/dask/array/tests/test_routines.py
+++ b/dask/array/tests/test_routines.py
@@ -1,6 +1,7 @@
 from __future__ import division, print_function, absolute_import
 
 import itertools
+import textwrap
 
 import pytest
 from distutils.version import LooseVersion
@@ -890,6 +891,8 @@ def test_choose():
 
 
 def test_piecewise():
+    np.random.seed(1337)
+
     x = np.random.randint(10, size=(15, 16))
     d = da.from_array(x, chunks=(4, 5))
 
@@ -898,16 +901,33 @@ def test_piecewise():
         da.piecewise(d, [d < 5, d >= 5], [lambda e, v, k: e + 1, 5], 1, k=2)
     )
 
+
+@pytest.mark.skipif(
+    LooseVersion(np.__version__) < '1.12.0',
+    reason=textwrap.dedent(
+        """\
+            NumPy piecewise mishandles the otherwise condition pre-1.12.0.
+
+            xref: https://github.com/numpy/numpy/issues/5737
+        """
+    )
+)
+def test_piecewise_otherwise():
+    np.random.seed(1337)
+
+    x = np.random.randint(10, size=(15, 16))
+    d = da.from_array(x, chunks=(4, 5))
+
     assert_eq(
         np.piecewise(
             x,
-            [x > 2, x <= 5],
+            [x > 5, x <= 2],
             [lambda e, v, k: e + 1, lambda e, v, k: v * e, lambda e, v, k: 0],
             1, k=2
         ),
         da.piecewise(
             d,
-            [d > 5, d <= 5],
+            [d > 5, d <= 2],
             [lambda e, v, k: e + 1, lambda e, v, k: v * e, lambda e, v, k: 0],
             1, k=2
         )
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index a7d15a898..fc9b5c2f9 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -3042,3 +3042,24 @@ def test_mixed_dask_array_multi_dimensional():
     assert_eq(ddf + dx + 1, df + x + 1)
     assert_eq(ddf + dx.rechunk((None, 1)) + 1, df + x + 1)
     assert_eq(ddf[['y', 'x']] + dx + 1, df[['y', 'x']] + x + 1)
+
+
+def test_meta_raises():
+    # Raise when we use a user defined fucntion
+    s = pd.Series(['abcd', 'abcd'])
+    ds = dd.from_pandas(s, npartitions=2)
+    try:
+        ds.map(lambda x: x[3])
+    except ValueError as e:
+        assert "meta=" in str(e)
+
+    # But not otherwise
+    df = pd.DataFrame({'a': ['x', 'y', 'y'],
+                       'b': ['x', 'y', 'z'],
+                       'c': [1, 2, 3]})
+    ddf = dd.from_pandas(df, npartitions=1)
+
+    with pytest.raises(Exception) as info:
+        ddf.a + ddf.c
+
+    assert "meta=" not in str(info.value)