[BUG] dataframe.apply cannot pickle 'weakref.ReferenceType' object

[hoarjour]

Describe the bug

When I try to use dataframe.apply like this:

df = md.DataFrame([[4, 9]] * 3, columns=['A', 'B'])
df = df.apply(lambda x: md.Series([1, 2], index=['foo', 'bar']), axis=1)
print(df.execute())

It will crash. But when I try:

df = md.DataFrame([[4, 9]] * 3, columns=['A', 'B'])
df = df.apply(lambda x: pd.Series([1, 2], index=['foo', 'bar']), axis=1)
print(df.execute())

It work well

To Reproduce

To help us reproducing this bug, please provide information below:

1. Your Python version: 3.10.6
2. The version of Mars you use: latest
3. Versions of crucial packages, such as numpy, scipy and pandas: follow mars
4. Full stack of the error.

Traceback (most recent call last):
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/subtask/worker/tests/collect.py", line 11, in <module>
    print(df.execute())
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/core/entity/tileables.py", line 404, in execute
    result = self.data.execute(session=session, **kw)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/core/entity/executable.py", line 148, in execute
    return execute(self, session=session, **kw)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/deploy/oscar/session.py", line 1875, in execute
    return session.execute(
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/deploy/oscar/session.py", line 1669, in execute
    execution_info: ExecutionInfo = fut.result(
  File "/Users/huanghaojie/miniconda3/envs/plotly/lib/python3.10/concurrent/futures/_base.py", line 458, in result
    return self.__get_result()
  File "/Users/huanghaojie/miniconda3/envs/plotly/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/deploy/oscar/session.py", line 1855, in _execute
    await execution_info
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/deploy/oscar/session.py", line 106, in wait
    return await self._aio_task
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/deploy/oscar/session.py", line 954, in _run_in_background
    raise task_result.error.with_traceback(task_result.traceback)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/task/supervisor/processor.py", line 373, in run
    await self._process_stage_chunk_graph(*stage_args)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/task/supervisor/processor.py", line 250, in _process_stage_chunk_graph
    chunk_to_result = await self._executor.execute_subtask_graph(
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/task/execution/mars/executor.py", line 200, in execute_subtask_graph
    return await stage_processor.run()
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/task/execution/mars/stage.py", line 231, in run
    return await self._run()
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/task/execution/mars/stage.py", line 251, in _run
    raise self.result.error.with_traceback(self.result.traceback)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/scheduling/worker/execution.py", line 392, in internal_run_subtask
    subtask_info.result = await self._retry_run_subtask(
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/scheduling/worker/execution.py", line 497, in _retry_run_subtask
    return await _retry_run(subtask, subtask_info, _run_subtask_once)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/scheduling/worker/execution.py", line 107, in _retry_run
    raise ex
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/scheduling/worker/execution.py", line 69, in _retry_run
    return await target_async_func(*args)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/scheduling/worker/execution.py", line 439, in _run_subtask_once
    return await asyncio.shield(aiotask)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/subtask/api.py", line 68, in run_subtask_in_slot
    return await ref.run_subtask.options(profiling_context=profiling_context).send(
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/context.py", line 196, in send
    return self._process_result_message(result)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/context.py", line 76, in _process_result_message
    raise message.as_instanceof_cause()
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/pool.py", line 584, in send
    result = await self._run_coro(message.message_id, coro)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/pool.py", line 341, in _run_coro
    return await coro
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/api.py", line 120, in __on_receive__
    return await super().__on_receive__(message)
  File "mars/oscar/core.pyx", line 523, in __on_receive__
  File "mars/oscar/core.pyx", line 516, in mars.oscar.core._BaseActor.__on_receive__
  File "mars/oscar/core.pyx", line 401, in _handle_actor_result
  File "mars/oscar/core.pyx", line 444, in _run_actor_async_generator
  File "mars/oscar/core.pyx", line 445, in mars.oscar.core._BaseActor._run_actor_async_generator
  File "mars/oscar/core.pyx", line 450, in mars.oscar.core._BaseActor._run_actor_async_generator
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/subtask/worker/runner.py", line 125, in run_subtask
    result = yield self._running_processor.run(subtask)
  File "mars/oscar/core.pyx", line 455, in mars.oscar.core._BaseActor._run_actor_async_generator
  File "mars/oscar/core.pyx", line 375, in _handle_actor_result
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/context.py", line 196, in send
    return self._process_result_message(result)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/context.py", line 76, in _process_result_message
    raise message.as_instanceof_cause()
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/pool.py", line 584, in send
    result = await self._run_coro(message.message_id, coro)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/backends/pool.py", line 341, in _run_coro
    return await coro
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/api.py", line 120, in __on_receive__
    return await super().__on_receive__(message)
  File "mars/oscar/core.pyx", line 523, in __on_receive__
  File "mars/oscar/core.pyx", line 516, in mars.oscar.core._BaseActor.__on_receive__
  File "mars/oscar/core.pyx", line 401, in _handle_actor_result
  File "mars/oscar/core.pyx", line 444, in _run_actor_async_generator
  File "mars/oscar/core.pyx", line 445, in mars.oscar.core._BaseActor._run_actor_async_generator
  File "mars/oscar/core.pyx", line 450, in mars.oscar.core._BaseActor._run_actor_async_generator
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/subtask/worker/processor.py", line 639, in run
    result = yield self._running_aio_task
  File "mars/oscar/core.pyx", line 455, in mars.oscar.core._BaseActor._run_actor_async_generator
  File "mars/oscar/core.pyx", line 375, in _handle_actor_result
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/subtask/worker/processor.py", line 484, in run
    ) = await self._store_data(chunk_graph)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/subtask/worker/processor.py", line 314, in _store_data
    store_infos = await put_infos
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/batch.py", line 146, in _async_batch
    return [await self._async_call(*args_list[0], **kwargs_list[0])]
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/batch.py", line 95, in _async_call
    return await self.func(*args, **kwargs)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/storage/api/oscar.py", line 107, in put
    return await self._storage_handler_ref.put(
  File "mars/oscar/core.pyx", line 263, in __pyx_actor_method_wrapper
  File "mars/oscar/core.pyx", line 266, in mars.oscar.core.__pyx_actor_method_wrapper
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/oscar/batch.py", line 95, in _async_call
    return await self.func(*args, **kwargs)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/services/storage/handler.py", line 172, in put
    object_info = await self._clients[level].put(obj)
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/storage/shared_memory.py", line 163, in put
    buffers = await serializer.run()
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/serialization/aio.py", line 85, in run
    return await self._get_buffers()
  File "/Users/huanghaojie/Documents/git_repo/mars/mars/serialization/aio.py", line 40, in _get_buffers
    headers, buffers = await serialize_with_spawn(
  File "mars/serialization/core.pyx", line 765, in serialize_with_spawn
  File "mars/serialization/core.pyx", line 605, in mars.serialization.core._serial_single
  File "mars/serialization/core.pyx", line 263, in mars.serialization.core.PickleSerializer.serial
  File "mars/serialization/core.pyx", line 228, in mars.serialization.core.pickle_buffers
  File "/Users/huanghaojie/miniconda3/envs/plotly/lib/python3.10/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/Users/huanghaojie/miniconda3/envs/plotly/lib/python3.10/site-packages/cloudpickle/cloudpickle_fast.py", line 632, in dump
    return Pickler.dump(self, obj)
TypeError: [address=127.0.0.1:52528, pid=5299] cannot pickle 'weakref.ReferenceType' object

6. Minimized code to reproduce the error.

Expected behavior

A clear and concise description of what you expected to happen.

Additional context

Add any other context about the problem here.