code: def to_ray_dataset(df, num_shards: int = None):
      """Create a Ray Dataset 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 Ray Dataset. 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 Ray Dataset
      """
      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
      chunk_refs: List["ray.ObjectRef"] = get_chunk_refs(df)
      dataset = ray_dataset.from_pandas_refs(chunk_refs)
      # 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(type(dataset), "__getstate__"):
          # if `dataframe` is not serializable by ray, patch our implementation
          dataset.__getstate__ = __getstate__
      return dataset
  
imports: import operator
  from functools import reduce
  from ....utils import lazy_import
  from .mldataset import _rechunk_if_needed
  from typing import Dict, List
label_desc: Use image node_id 1 for calling a global function with example usage: to_ray_dataset(df, num_shards) and returns: dataset, state