dask fixes

Signed-off-by: Anatoly Myachev <[email protected]>

modin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py

@@ -41,3 +41,24 @@ class PandasOnDaskDataframe(PandasDataframe):
     """
 
     _partition_mgr_cls = PandasOnDaskDataframePartitionManager
+
+    @classmethod
+    def reconnect(cls, address, attributes):
+        try:
+            from distributed import default_client
+
+            default_client()
+        except ValueError:
+            from distributed import Client
+
+            # setup `default_client` for worker process
+            _ = Client(address)
+        obj = cls.__new__(cls)
+        obj.__dict__.update(attributes)
+        return obj
+
+    def __reduce__(self):
+        from distributed import default_client
+
+        address = default_client().scheduler_info()["address"]
+        return self.reconnect, (address, self.__dict__)

modin/pandas/test/test_series.py

@@ -4796,7 +4796,7 @@ def test_binary_numpy_universal_function_issue_6483():
     )
 
 
-@pytest.mark.xfail(Engine.get() != "Ray", reason="reason")
+@pytest.mark.skipif(Engine.get() == "Unidist", reason="Unidist doesn't work for now.")
 def test__reduce__():
     abbreviations = pd.Series(
         ["Major League Baseball", "National Basketball Association"],