use ddf no_default instead of pd_no_default

pandas-dev/pandas#40397 (comment)

Author: gjoseph92

dask/dataframe/core.py

@@ -16,7 +16,6 @@
     is_numeric_dtype,
     is_datetime64_any_dtype,
 )
-from pandas._libs.lib import no_default as pd_no_default
 from tlz import merge, first, unique, partition_all, remove
 
 try:
@@ -1417,9 +1416,7 @@ def replace(self, to_replace=None, value=None, regex=False):
             enforce_metadata=False,
         )
 
-    def to_dask_array(
-        self, lengths=None, dtype=None, na_value=pd_no_default, meta=None
-    ):
+    def to_dask_array(self, lengths=None, dtype=None, na_value=no_default, meta=None):
         """Convert a dask DataFrame to a dask array.
 
         Parameters
@@ -1452,7 +1449,7 @@ def to_dask_array(
         Returns
         -------
         """
-        if not PANDAS_GT_110 and na_value is not pd_no_default:
+        if not PANDAS_GT_110 and na_value is not no_default:
             raise NotImplementedError(
                 "na_value is not a valid argument for to_dask_array"
                 f"if pandas < 1.1.0. Pandas version is {pd.__version__}"

dask/dataframe/methods.py

@@ -3,7 +3,7 @@
 import numpy as np
 import pandas as pd
 from pandas.api.types import is_categorical_dtype, union_categoricals
-from pandas._libs.lib import no_default as pd_no_default
+from pandas.api.extensions import no_default as pd_no_default
 from tlz import partition
 
 from ._compat import PANDAS_GT_110
@@ -326,7 +326,19 @@ def size(x):
     return x.size
 
 
-def to_numpy(df, _dtype, na_value=pd_no_default):
+def to_numpy(df, _dtype, na_value):
+    # callers use the "__no_default__" string instead of `pd_no_default`,
+    # since `pd_no_default` isn't consistently pickleable or tokenizeable
+    # https://github.com/pandas-dev/pandas/issues/40397
+    from .core import no_default
+
+    try:
+        if na_value == no_default:  # `==` not `is`, since it could be pickled
+            na_value = pd_no_default
+    except Exception:
+        # could be pd.NA, where __bool__ will raise an error
+        pass
+
     kwargs = {"dtype": _dtype, "copy": False}
     if PANDAS_GT_110:
         kwargs["na_value"] = na_value