Repository: dask/dask
Pull request: Option to not check meta while using from_delayed
closes #5435

Problem statement:
ValueError: Metadata mismatch found in `from_delayed`.
It seems that merging multiple delayed dataframes does not work when the dataframes are series of category type.

It seems that the error is coming from the equality of categorical dtypes includes the categories used, which makes this approach to concatenation not work. This came up in the context of intake.

```python
from pandas import DataFrame, Series
from dask.delayed import delayed
from dask.dataframe import from_delayed

a = Series(['a', 'b', 'a'], dtype='category')
b = Series(['a', 'c', 'a'], dtype='category')
da = delayed(lambda x: x)(a)
db = delayed(lambda x: x)(b)

c = from_delayed([da, db])
c.compute()
```
Error msg
```python
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-17-02c1dd8d067d> in <module>
----> 1 c.compute()

~/mc/lib/python3.7/site-packages/dask/base.py in compute(self, **kwargs)
    173         dask.base.compute
    174         """
--> 175         (result,) = compute(self, traverse=False, **kwargs)
    176         return result
    177 

~/mc/lib/python3.7/site-packages/dask/base.py in compute(*args, **kwargs)
    444     keys = [x.__dask_keys__() for x in collections]
    445     postcomputes = [x.__dask_postcompute__() for x in collections]
--> 446     results = schedule(dsk, keys, **kwargs)
    447     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    448 

~/mc/lib/python3.7/site-packages/dask/threaded.py in get(dsk, result, cache, num_workers, pool, **kwargs)
     80         get_id=_thread_get_id,
     81         pack_exception=pack_exception,
---> 82         **kwargs
     83     )
     84 

~/mc/lib/python3.7/site-packages/dask/local.py in get_async(apply_async, num_workers, dsk, result, cache, get_id, rerun_exceptions_locally, pack_exception, raise_exception, callbacks, dumps, loads, **kwargs)
    489                         _execute_task(task, data)  # Re-execute locally
    490                     else:
--> 491                         raise_exception(exc, tb)
    492                 res, worker_id = loads(res_info)
    493                 state["cache"][key] = res

~/mc/lib/python3.7/site-packages/dask/compatibility.py in reraise(exc, tb)
    128         if exc.__traceback__ is not tb:
    129             raise exc.with_traceback(tb)
--> 130         raise exc
    131 
    132     import pickle as cPickle

~/mc/lib/python3.7/site-packages/dask/local.py in execute_task(key, task_info, dumps, loads, get_id, pack_exception)
    231     try:
    232         task, data = loads(task_info)
--> 233         result = _execute_task(task, data)
    234         id = get_id()
    235         result = dumps((result, id))

~/mc/lib/python3.7/site-packages/dask/core.py in _execute_task(arg, cache, dsk)
    117         func, args = arg[0], arg[1:]
    118         args2 = [_execute_task(a, cache) for a in args]
--> 119         return func(*args2)
    120     elif not ishashable(arg):
    121         return arg

~/mc/lib/python3.7/site-packages/dask/dataframe/utils.py in check_meta(x, meta, funcname, numeric_equal)
    650     raise ValueError(
    651         "Metadata mismatch found%s.\n\n"
--> 652         "%s" % ((" in `%s`" % funcname if funcname else ""), errmsg)
    653     )
    654 

ValueError: Metadata mismatch found in `from_delayed`.

Partition type: `pandas.core.series.Series`
+----------+----------+
|          | dtype    |
+----------+----------+
| Found    | category |
| Expected | category |
+----------+----------+
```

Hints from issue discussion:
Thank you for the excellent issue @CJ-Wright .  

Looking into the code a bit it looks like this strictness in `check_meta` was intentional at some point:

https://github.com/dask/dask/blob/8001ee2e258f3f27d40e78990c5b98d44edc138e/dask/dataframe/utils.py#L593-L602

And it looks like we enforce the metadata check in `from_delayed` here:

https://github.com/dask/dask/blob/8001ee2e258f3f27d40e78990c5b98d44edc138e/dask/dataframe/io/io.py#L565-L570

So one answer to this issue is "make sure that all of your dataframes are consistent with each other".  

However, many Dask Dataframe algorithms *are* robust to mismatched categoricals.  Probably in this case you would want to tell Dask dataframe that you're not sure what the categories of your dataframe are. The `dd.from_delayed` function takes a `meta=` keyword, where you can provide metadata for your full dataframe.  If you don't provide something then it computes the first partition and uses that.  You might provide metadata, but with explicitly unknown categories.  It looks like there is a function in `dd.utils.clear_known_categories` that might be helpful?  (I actually don't know this part of the code at all, so I'm just guessing).

Hopefully some of that provides context here.  If you have suggestions on how this can be fixed/improved generally, that would also be welcome.
Hmm ok. Would it be possible to have a boolean flag on `from_delayed` which says don't check the meta? I tried using the `clear_known_categories` function, but that got rid of my categories all together.
Yes, that seems reasonable

On Thu, Sep 26, 2019, 12:12 PM Christopher J. Wright <
notifications@github.com> wrote:

> Hmm ok. Would it be possible to have a boolean flag on from_delayed which
> says don't check the meta? I tried using the clear_known_categories
> function, but that got rid of my categories all together.
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/issues/5435?email_source=notifications&email_token=AACKZTEIAJHLCZSNIBSBAKTQLTUQDA5CNFSM4I24QEUKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD7WJYZA#issuecomment-535600228>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AACKZTHWT5K6DDR6XPCIGBLQLTUQDANCNFSM4I24QEUA>
> .
>

Would you like a PR to that effect?
Sure

On Thu, Sep 26, 2019 at 12:15 PM Christopher J. Wright <
notifications@github.com> wrote:

> Would you like a PR to that effect?
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/issues/5435?email_source=notifications&email_token=AACKZTB34LEH5VVXHTBNCS3QLTU3JA5CNFSM4I24QEUKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD7WKCBI#issuecomment-535601413>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AACKZTC5LLESCH3N2AHX54DQLTU3JANCNFSM4I24QEUA>
> .
>

Commit message: add test

Patch to review:
diff --git a/dask/dataframe/io/io.py b/dask/dataframe/io/io.py
index d1c47f96455..675d9c089ab 100644
--- a/dask/dataframe/io/io.py
+++ b/dask/dataframe/io/io.py
@@ -524,7 +524,8 @@ def to_records(df):
 
 
 @insert_meta_param_description
-def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed"):
+def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed", 
+                 check_meta_flag=True):
     """ Create Dask DataFrame from many Dask Delayed objects
 
     Parameters
@@ -542,6 +543,9 @@ def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed"):
         If None, then won't use index information
     prefix : str, optional
         Prefix to prepend to the keys.
+    check_meta_flag : bool, optional
+        If False don't check that the incoming dataframe metadata are the 
+        same as the outbound metadata, defaults to True
     """
     from dask.delayed import Delayed
 
@@ -562,12 +566,20 @@ def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed"):
 
     name = prefix + "-" + tokenize(*dfs)
     dsk = merge(df.dask for df in dfs)
-    dsk.update(
-        {
-            (name, i): (check_meta, df.key, meta, "from_delayed")
-            for (i, df) in enumerate(dfs)
-        }
-    )
+    if check_meta_flag:
+        dsk.update(
+            {
+                (name, i): (check_meta, df.key, meta, "from_delayed")
+                for (i, df) in enumerate(dfs)
+            }
+        )
+    else:
+        dsk.update(
+            {
+                (name, i): (lambda x, *_: x, df.key, meta, "from_delayed")
+                for (i, df) in enumerate(dfs)
+            }
+        )
 
     if divisions is None or divisions == "sorted":
         divs = [None] * (len(dfs) + 1)
diff --git a/dask/tests/test_delayed.py b/dask/tests/test_delayed.py
index dac5a5264d9..36b95d31ecd 100644
--- a/dask/tests/test_delayed.py
+++ b/dask/tests/test_delayed.py
@@ -600,3 +600,17 @@ def test_attribute_of_attribute():
     assert isinstance(x.a, Delayed)
     assert isinstance(x.a.b, Delayed)
     assert isinstance(x.a.b.c, Delayed)
+
+
+def test_check_meta_flag():
+    from pandas import DataFrame, Series
+    from dask.delayed import delayed
+    from dask.dataframe import from_delayed
+
+    a = Series(['a', 'b', 'a'], dtype='category')
+    b = Series(['a', 'c', 'a'], dtype='category')
+    da = delayed(lambda x: x)(a)
+    db = delayed(lambda x: x)(b)
+
+    c = from_delayed([da, db])
+    c.compute()


Code review comments:
diff_hunk: @@ -562,12 +566,20 @@ def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed"):
 
     name = prefix + "-" + tokenize(*dfs)
     dsk = merge(df.dask for df in dfs)
-    dsk.update(
-        {
-            (name, i): (check_meta, df.key, meta, "from_delayed")
-            for (i, df) in enumerate(dfs)
-        }
-    )
+    if check_meta_flag:
+        dsk.update(
+            {
+                (name, i): (check_meta, df.key, meta, "from_delayed")
+                for (i, df) in enumerate(dfs)
+            }
+        )
+    else:
+        dsk.update(
+            {
+                (name, i): (lambda x, *_: x, df.key, meta, "from_delayed")
path: dask/dataframe/io/io.py
text: @user1:
This lambda seems to be a no-op.  Maybe insteaad something like the following?

```suggestion
                (name, i): df.key,
```

diff_hunk: @@ -542,6 +543,9 @@ def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed"):
         If None, then won't use index information
     prefix : str, optional
         Prefix to prepend to the keys.
+    check_meta_flag : bool, optional
+        If False don't check that the incoming dataframe metadata are the 
+        same as the outbound metadata, defaults to True
path: dask/dataframe/io/io.py
text: @user1:
The terms inbound and outbound aren't immediately clear to me here.

How about "Check that the partitions have consistent metadata"

diff_hunk: @@ -524,7 +524,8 @@ def to_records(df):
 
 
 @insert_meta_param_description
-def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed"):
+def from_delayed(dfs, meta=None, divisions=None, prefix="from-delayed", 
+                 check_meta_flag=True):
path: dask/dataframe/io/io.py
text: @user1:
How about `check_meta` or `verify_meta` instead?  The suffix `_flag` might not be necessary here.

@author:
I think `check_meta` will bump into the function `check_meta`, but `verify_meta` works for me.

diff_hunk: @@ -600,3 +600,17 @@ def test_attribute_of_attribute():
     assert isinstance(x.a, Delayed)
     assert isinstance(x.a.b, Delayed)
     assert isinstance(x.a.b.c, Delayed)
+
+
+def test_check_meta_flag():
+    from pandas import DataFrame, Series
+    from dask.delayed import delayed
+    from dask.dataframe import from_delayed
+
+    a = Series(['a', 'b', 'a'], dtype='category')
+    b = Series(['a', 'c', 'a'], dtype='category')
+    da = delayed(lambda x: x)(a)
+    db = delayed(lambda x: x)(b)
+
+    c = from_delayed([da, db])
+    c.compute()
path: dask/tests/test_delayed.py
text: @user1:
We try to use `assert_eq` rather than compute in tests when possible

```suggestion
    assert_eq(c, c)
```