code: def spawn(
      func,
      args=(),
      kwargs=None,
      retry_when_fail=False,
      resolve_tileable_input=False,
      n_output=None,
      **kw,
  ):
      """
      Spawn a function and return a Mars Object which can be executed later.
  
      Parameters
      ----------
      func : function
          Function to spawn.
      args: tuple
         Args to pass to function
      kwargs: dict
         Kwargs to pass to function
      retry_when_fail: bool, default False
         If True, retry when function failed.
      resolve_tileable_input: bool default False
         If True, resolve tileable inputs as values.
      n_output: int
         Count of outputs for the function
  
      Returns
      -------
      Object
          Mars Object.
  
      Examples
      --------
      >>> import mars.remote as mr
      >>> def inc(x):
      >>>     return x + 1
      >>>
      >>> result = mr.spawn(inc, args=(0,))
      >>> result
      Object <op=RemoteFunction, key=e0b31261d70dd9b1e00da469666d72d9>
      >>> result.execute().fetch()
      1
  
      List of spawned functions can be converted to :class:`mars.remote.ExecutableTuple`,
      and `.execute()` can be called to run together.
  
      >>> results = [mr.spawn(inc, args=(i,)) for i in range(10)]
      >>> mr.ExecutableTuple(results).execute().fetch()
      [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
  
      Mars Object returned by :meth:`mars.remote.spawn` can be treated
      as arguments for other spawn functions.
  
      >>> results = [mr.spawn(inc, args=(i,)) for i in range(10)]   # list of spawned functions
      >>> def sum_all(xs):
              return sum(xs)
      >>> mr.spawn(sum_all, args=(results,)).execute().fetch()
      55
  
      inside a spawned function, new functions can be spawned.
  
      >>> def driver():
      >>>     results = [mr.spawn(inc, args=(i,)) for i in range(10)]
      >>>     return mr.ExecutableTuple(results).execute().fetch()
      >>>
      >>> mr.spawn(driver).execute().fetch()
      [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
  
      Mars tensor, DataFrame and so forth is available in spawned functions as well.
  
      >>> import mars.tensor as mt
      >>> def driver2():
      >>>     t = mt.random.rand(10, 10)
      >>>     return t.sum().to_numpy()
      >>>
      >>> mr.spawn(driver2).execute().fetch()
      52.47844223908132
  
      Argument of `n_output` can indicate that the spawned function will return multiple outputs.
      This is important when some of the outputs may be passed to different functions.
  
      >>> def triage(alist):
      >>>     ret = [], []
      >>>     for i in alist:
      >>>         if i < 0.5:
      >>>             ret[0].append(i)
      >>>         else:
      >>>             ret[1].append(i)
      >>>     return ret
      >>>
      >>> def sum_all(xs):
      >>>     return sum(xs)
      >>>
      >>> l = [0.4, 0.7, 0.2, 0.8]
      >>> la, lb = mr.spawn(triage, args=(l,), n_output=2)
      >>>
      >>> sa = mr.spawn(sum_all, args=(la,))
      >>> sb = mr.spawn(sum_all, args=(lb,))
      >>> mr.ExecutableTuple([sa, sb]).execute().fetch()
      >>> [0.6000000000000001, 1.5]
      """
      if not isinstance(args, tuple):
          args = [args]
      else:
          args = list(args)
      if kwargs is None:
          kwargs = dict()
      if not isinstance(kwargs, dict):
          raise TypeError("kwargs has to be a dict")
  
      op = RemoteFunction(
          function=func,
          function_args=args,
          function_kwargs=kwargs,
          retry_when_fail=retry_when_fail,
          resolve_tileable_input=resolve_tileable_input,
          n_output=n_output,
          **kw,
      )
      if op.extra_params:
          raise ValueError(f"Unexpected kw: {list(op.extra_params)[0]}")
      return op()
  
imports: from collections import UserDict
  from collections.abc import Iterable
  from functools import partial
  from ..None import opcodes
  from ..core import ENTITY_TYPE, ChunkData, Tileable
  from ..core.custom_log import redirect_custom_log
  from ..core.operand import ObjectOperand
  from ..dataframe.core import DATAFRAME_TYPE, SERIES_TYPE, INDEX_TYPE
  from ..serialization.serializables import FunctionField, ListField, DictField, BoolField, Int32Field
  from ..tensor.core import TENSOR_TYPE
  from ..utils import build_fetch_tileable, enter_current_session, find_objects, replace_objects, merge_chunks, merged_chunk_as_tileable_type
  from .operands import RemoteOperandMixin
label_desc: Use image node_id 1 for calling a global function with example usage: spawn(func, args, kwargs, retry_when_fail, resolve_tileable_input, n_output) and returns: op