code: def to_ray_mldataset(df, num_shards: int = None):
      """Create a MLDataset from Mars DataFrame
  
      Args:
          df (mars.dataframe.Dataframe): the Mars DataFrame
          num_shards (int, optional): the number of shards that will be created
              for the MLDataset. Defaults to None.
              If num_shards is None, chunks will be grouped by nodes where they lie.
              Otherwise, chunks will be grouped by their order in DataFrame.
  
      Returns:
          a MLDataset
      """
      df = _rechunk_if_needed(df, num_shards)
      # chunk_addr_refs is fetched directly rather than in batches
      # during `fetch` procedure, it'll be checked that df has been executed
      # items in chunk_addr_refs are ordered by positions in df
      # while adjacent chunks may belong to different addrs, i.e.
      #       chunk1 for addr1,
      #       chunk2 & chunk3 for addr2,
      #       chunk4 for addr1
      fetched_infos: Dict[str, List] = df.fetch_infos(
          fields=["bands", "object_refs"]
      )
      chunk_addr_refs: List[Tuple[Tuple, "ray.ObjectRef"]] = []
      for bands, object_refs in zip(
          fetched_infos["bands"], fetched_infos["object_refs"]
      ):
          chunk_addr_ref = (
              (bands[0], object_refs[0])
              if bands
              else ("ray_dag_0", object_refs[0])
          )
          chunk_addr_refs.append(chunk_addr_ref)
  
      group_to_obj_refs: Dict[
          str, List[ray.ObjectRef]
      ] = _group_chunk_refs(chunk_addr_refs, num_shards)
  
      record_batches = []
      for rank, obj_refs in enumerate(group_to_obj_refs.values()):
          record_batches.append(
              ChunkRefBatch(shard_id=rank, obj_refs=obj_refs)
          )
      worker_cls = ray.remote(num_cpus=0)(
          parallel_it.ParallelIteratorWorker
      )
      actors = [worker_cls.remote(g, False) for g in record_batches]
      it = parallel_it.from_actors(actors, "from_mars")
      dataset = ml_dataset.from_parallel_iter(
          it, need_convert=False, batch_size=0
      )
      # Hold mars dataframe to avoid mars dataframe and ray object gc.
      dataset.dataframe = df
  
      def __getstate__():
          state = dataset.__dict__.copy()
          state.pop("dataframe", None)
          return state
  
      if not hasattr(dataset, "__getstate__"):
          # `dataframe` is not serializable by ray.
          dataset.__getstate__ = __getstate__
      return dataset
  
file_path: mars/mars/dataframe/contrib/raydataset/mldataset.py
imports: import numpy
  import pandas
  from collections import defaultdict
  from typing import Dict, Iterable, List, Tuple
  from ....utils import lazy_import
label_desc: Use image node_id 3 for calling a global function with example usage: to_ray_mldataset(df, num_shards) and returns: dataset, state