[BACKPORT] [Ray] Fix serializing lambdas in web (#2512)

mars/deploy/oscar/tests/test_local.py

@@ -214,10 +214,6 @@ async def test_sync_execute_in_async(create_cluster):
     np.testing.assert_array_equal(res, np.ones((10, 10)) + 1)
 
 
-def _my_func():
-    print('output from function')
-
-
 async def _run_web_session_test(web_address):
     session_id = str(uuid.uuid4())
     session = await AsyncSession.init(web_address, session_id)
@@ -235,6 +231,10 @@ async def _run_web_session_test(web_address):
     np.testing.assert_equal(raw + 1, await session.fetch(b))
     del a, b
 
+    # Test spawn a local function by the web session.
+    def _my_func():
+        print('output from function')
+
     r = mr.spawn(_my_func)
     info = await session.execute(r)
     await info
@@ -249,6 +249,16 @@ async def _run_web_session_test(web_address):
                                                      offsets={r.op.key: '0k'},
                                                      sizes=[1000]))
 
+    df = md.DataFrame([1, 2, 3])
+    # Test apply a lambda by the web session.
+    r = df.apply(lambda x: x)
+    info = await session.execute(r)
+    await info
+    assert info.result() is None
+    assert info.exception() is None
+    assert info.progress() == 1
+    pd.testing.assert_frame_equal(await session.fetch(r), pd.DataFrame([1, 2, 3]))
+
     AsyncSession.reset_default()
     await session.destroy()
 

mars/tests/test_utils.py

@@ -35,9 +35,12 @@
     pd = None
 import pytest
 
+from .. import dataframe as md
 from .. import tensor as mt
 from .. import utils
-from ..core import tile
+from ..core import tile, TileableGraph
+from ..serialization.ray import register_ray_serializers
+from .core import require_ray
 
 
 def test_string_conversion():
@@ -415,3 +418,16 @@ def test_readable_size():
     assert utils.readable_size(14354000) == '13.69M'
     assert utils.readable_size(14354000000) == '13.37G'
     assert utils.readable_size(14354000000000) == '13.05T'
+
+
+@require_ray
+def test_web_serialize_lambda():
+    register_ray_serializers()
+    df = md.DataFrame(
+            mt.random.rand(10_0000, 4, chunk_size=1_0000),
+            columns=list('abcd'))
+    r = df.apply(lambda x: x)
+    graph = TileableGraph([r])
+    s = utils.serialize_serializable(graph)
+    f = utils.deserialize_serializable(s)
+    assert isinstance(f, TileableGraph)

mars/utils.py

@@ -24,7 +24,7 @@
 import numbers
 import operator
 import os
-import pickle
+import cloudpickle as pickle
 import pkgutil
 import random
 import shutil