Repository: dask/dask
Pull request: Implement dask.dataframe.to_numeric
Closes: #5698 

- [x] Tests added / passed
- [x] Passes `black dask` / `flake8 dask`

@TomAugspurger I haven't added any of the kwargs yet, but just wanted to make sure that this is what you intended.

Problem statement:
Implement dask.dataframe.to_numeric
Which mimics `pandas.to_numeric`.

For scalar inputs, this could return a Delayed object.

For array inputs, this would return a `dask.array.Array` and would be `.map_blocks(pd.to_numeric)`.

For Series, it'd be `.map_partitions(pd.to_numeric)`.

I'm not sure if downcasting can be easily supported, since we'd need to scan the data for min / max values.

Hints from issue discussion:
Just to say that I am taking a crack at this

Commit message: Implement downcast and errors kwargs

Patch to review:
diff --git a/dask/dataframe/__init__.py b/dask/dataframe/__init__.py
index 064a795b9df..a7ffeb53b68 100644
--- a/dask/dataframe/__init__.py
+++ b/dask/dataframe/__init__.py
@@ -29,6 +29,7 @@
         to_json,
         read_fwf,
     )
+    from .numeric import to_numeric
     from .optimize import optimize
     from .multi import merge, concat, merge_asof
     from . import rolling, backends
diff --git a/dask/dataframe/numeric.py b/dask/dataframe/numeric.py
new file mode 100644
index 00000000000..0f58ccfdb10
--- /dev/null
+++ b/dask/dataframe/numeric.py
@@ -0,0 +1,70 @@
+import pandas as pd
+from pandas.core.dtypes.common import is_scalar as pd_is_scalar
+
+from ..utils import derived_from
+from ..delayed import delayed
+from ..array import Array
+from .core import Series
+
+
+__all__ = ("to_numeric",)
+
+
+@derived_from(pd)
+def to_numeric(arg, errors="raise", downcast=None, meta=None):
+    """
+    Return type depends on input. Delayed if scalar, otherwise same as input.
+    """
+    if downcast not in (None, "integer", "signed", "unsigned", "float"):
+        raise ValueError("invalid downcasting method provided")
+
+    if errors not in ("ignore", "raise", "coerce"):
+        raise ValueError("invalid error value specified")
+
+    is_series = isinstance(arg, Series)
+    is_array = isinstance(arg, Array)
+    is_scalar = pd_is_scalar(arg)
+
+    if not any([is_series, is_array, is_scalar]):
+        raise TypeError(
+            "arg must be a list, tuple, dask.array.Array, or dask.dataframe.Series"
+        )
+
+    if meta is not None:
+        if is_scalar:
+            raise KeyError("``meta`` is not allowed when input is a scalar.")
+        if downcast is not None:
+            raise KeyError("Only one of downcast and meta should be specified.")
+    else:
+        if is_series or is_array:
+            if downcast is None:
+                meta = pd.to_numeric(arg._meta)
+            else:
+                if downcast in ["integer", "signed"]:
+                    dtype = "int8"
+                elif downcast == "unsigned":
+                    dtype = "uint8"
+                elif downcast == "float":
+                    dtype = "float32"
+                meta = (0, dtype)
+
+    if is_series:
+        return arg.map_partitions(
+            pd.to_numeric,
+            token=arg._name + "-to_numeric",
+            meta=meta,
+            enforce_metadata=False,
+            errors=errors,
+            downcast=downcast,
+        )
+    if is_array:
+        return arg.map_blocks(
+            pd.to_numeric,
+            name=arg._name + "-to_numeric",
+            meta=meta,
+            errors=errors,
+            downcast=downcast,
+        )
+    if is_scalar:
+        obj = lambda x: pd.to_numeric(x, errors=errors, downcast=downcast)
+        return delayed(obj, name="to_numeric", pure=True)(arg)
diff --git a/dask/dataframe/tests/test_numeric.py b/dask/dataframe/tests/test_numeric.py
new file mode 100644
index 00000000000..ac3c3dca144
--- /dev/null
+++ b/dask/dataframe/tests/test_numeric.py
@@ -0,0 +1,74 @@
+import pytest
+import numpy as np
+import pandas as pd
+
+from dask.array import from_array, Array
+from dask.delayed import Delayed
+from dask.dataframe import from_pandas, Series, to_numeric
+
+
+@pytest.mark.parametrize("arg", ["5", 5, "5 "])
+def test_to_numeric_on_scalars(arg):
+    output = to_numeric(arg)
+    assert isinstance(output, Delayed)
+    assert output.compute() == 5
+
+
+def test_to_numeric_on_dask_array():
+    arg = from_array(["1.0", "2", -3, 5.1])
+    expected = np.array([1.0, 2.0, -3.0, 5.1])
+    output = to_numeric(arg)
+    assert isinstance(output, Array)
+    assert list(output.compute()) == list(expected)
+
+
+def test_to_numeric_on_dask_dataframe_series():
+    s = pd.Series(["1.0", "2", -3, -5.1])
+    arg = from_pandas(s, npartitions=2)
+    expected = pd.to_numeric(s)
+    output = to_numeric(arg)
+    assert output.dtype == "int64"
+    assert isinstance(output, Series)
+    assert list(output.compute()) == list(expected)
+
+
+def test_to_numeric_on_dask_dataframe_series_with_meta():
+    s = pd.Series(["1.0", "2", -3, -5.1])
+    arg = from_pandas(s, npartitions=2)
+    expected = pd.to_numeric(s)
+    output = to_numeric(arg, meta=pd.Series([], dtype="float64"))
+    assert output.dtype == "float64"
+    assert isinstance(output, Series)
+    assert list(output.compute()) == list(expected)
+
+
+def test_to_numeric_on_dask_dataframe_series_with_downcast_as_valid_integer():
+    s = pd.Series(["1", "2"])
+    arg = from_pandas(s, npartitions=2)
+    expected = pd.to_numeric(s)
+    output = to_numeric(arg, downcast="integer")
+    assert output.dtype == "int8"
+    assert isinstance(output, Series)
+    actual = output.compute()
+    assert actual.dtype == "int8"
+    assert list(actual) == list(expected)
+
+
+def test_to_numeric_on_dask_dataframe_series_with_downcast_as_invalid_integer():
+    s = pd.Series(["1", "2.3"])
+    arg = from_pandas(s, npartitions=2)
+    expected = pd.to_numeric(s)
+    output = to_numeric(arg, downcast="integer")
+    assert output.dtype == "int8"
+    assert isinstance(output, Series)
+    actual = output.compute()
+    assert actual.dtype == "float64"
+    assert list(actual) == list(expected)
+
+
+def test_to_numeric_on_dask_dataframe_dataframe_raises_error():
+    s = pd.Series(["1.0", "2", -3, -5.1])
+    df = pd.DataFrame({"a": s, "b": s})
+    arg = from_pandas(df, npartitions=2)
+    with pytest.raises(TypeError, match="arg must be a list, tuple, dask."):
+        to_numeric(arg)


Code review comments:
diff_hunk: @@ -0,0 +1,70 @@
+import pandas as pd
+from pandas.core.dtypes.common import is_scalar as pd_is_scalar
+
+from ..utils import derived_from
+from ..delayed import delayed
+from ..array import Array
+from .core import Series
+
+
+__all__ = ("to_numeric",)
+
+
+@derived_from(pd)
+def to_numeric(arg, errors="raise", downcast=None, meta=None):
+    """
+    Return type depends on input. Delayed if scalar, otherwise same as input.
+    """
+    if downcast not in (None, "integer", "signed", "unsigned", "float"):
+        raise ValueError("invalid downcasting method provided")
+
+    if errors not in ("ignore", "raise", "coerce"):
+        raise ValueError("invalid error value specified")
+
+    is_series = isinstance(arg, Series)
+    is_array = isinstance(arg, Array)
+    is_scalar = pd_is_scalar(arg)
+
+    if not any([is_series, is_array, is_scalar]):
+        raise TypeError(
+            "arg must be a list, tuple, dask.array.Array, or dask.dataframe.Series"
+        )
+
+    if meta is not None:
+        if is_scalar:
+            raise KeyError("``meta`` is not allowed when input is a scalar.")
+        if downcast is not None:
+            raise KeyError("Only one of downcast and meta should be specified.")
+    else:
+        if is_series or is_array:
+            if downcast is None:
+                meta = pd.to_numeric(arg._meta)
+            else:
+                if downcast in ["integer", "signed"]:
+                    dtype = "int8"
+                elif downcast == "unsigned":
+                    dtype = "uint8"
+                elif downcast == "float":
+                    dtype = "float32"
+                meta = (0, dtype)
+
+    if is_series:
+        return arg.map_partitions(
+            pd.to_numeric,
+            token=arg._name + "-to_numeric",
+            meta=meta,
+            enforce_metadata=False,
+            errors=errors,
+            downcast=downcast,
+        )
+    if is_array:
+        return arg.map_blocks(
+            pd.to_numeric,
+            name=arg._name + "-to_numeric",
+            meta=meta,
+            errors=errors,
+            downcast=downcast,
+        )
+    if is_scalar:
+        obj = lambda x: pd.to_numeric(x, errors=errors, downcast=downcast)
path: dask/dataframe/numeric.py
text: @user1:
We try to avoid `lambda`s in dask.dataframe. Does `delayed(pd.to_numeric)(arg, errors=errors, downcast=downcast)` work?

@author:
Ok! Thanks for the tip :)

diff_hunk: @@ -0,0 +1,70 @@
+import pandas as pd
+from pandas.core.dtypes.common import is_scalar as pd_is_scalar
+
+from ..utils import derived_from
+from ..delayed import delayed
+from ..array import Array
+from .core import Series
+
+
+__all__ = ("to_numeric",)
+
+
+@derived_from(pd)
+def to_numeric(arg, errors="raise", downcast=None, meta=None):
+    """
+    Return type depends on input. Delayed if scalar, otherwise same as input.
+    """
+    if downcast not in (None, "integer", "signed", "unsigned", "float"):
path: dask/dataframe/numeric.py
text: @user1:
Thoughts on checking these here vs. letting pandas catch issues when the graph is executed? I'm not sure if we're consistent.

The upside of checking them here is that the errors are caught earlier and the stacktrace is nicer. The downside is that if pandas adds a new option, we'll need to explicitly add it here.

@author:
Right, I was thinking that the early failure is worth the duplication, and there already need to be changes if pandas adds options.

diff_hunk: @@ -0,0 +1,70 @@
+import pandas as pd
+from pandas.core.dtypes.common import is_scalar as pd_is_scalar
path: dask/dataframe/numeric.py
text: @user1:
Is this available from `pd.api.types`? `pandas.core` is generally considered private.

@author:
Sure! I am not sure where I got that from. Probably pandas itself.