class_docstr: Reader object that loads data from Ray Dataset.
  
  Examples:
      config = {
          "input": "dataset",
          "input_config": {
              "format": "json",
              # A single data file, a directory, or anything
              # that ray.data.dataset recognizes.
              "paths": "/tmp/sample_batches/",
              # By default, parallelism=num_workers.
              "parallelism": 3,
              # Dataset allocates 0.5 CPU for each reader by default.
              # Adjust this value based on the size of your offline dataset.
              "num_cpus_per_read_task": 0.5,
          }
      }
docstr: Initializes a DatasetReader instance.
  
  Args:
      ds: Ray dataset to sample from.
docstr_tok: ["Initializes","a","DatasetReader","instance",".","Args",":","ds",":","Ray","dataset","to","sample","from","."]
code: def __init__(
      self, ds: ray.data.Dataset, ioctx: Optional[IOContext] = None
  ):
      """Initializes a DatasetReader instance.
  
      Args:
          ds: Ray dataset to sample from.
      """
      self._ioctx = ioctx or IOContext()
      self._default_policy = self.policy_map = None
      self.preprocessor = None
      self._dataset = ds
      self.count = None if not self._dataset else self._dataset.count()
      # do this to disable the ray data stdout logging
      ray.data.set_progress_bars(enabled=False)
  
      # the number of steps to return per call to next()
      self.batch_size = self._ioctx.config.get("train_batch_size", 1)
      num_workers = self._ioctx.config.get("num_workers", 0)
      seed = self._ioctx.config.get("seed", None)
      if num_workers:
          self.batch_size = max(
              math.ceil(self.batch_size / num_workers), 1
          )
      # We allow the creation of a non-functioning None DatasetReader.
      # It's useful for example for a non-rollout local worker.
      if ds:
          if self._ioctx.worker is not None:
              self._policy_map = self._ioctx.worker.policy_map
              self._default_policy = self._policy_map.get(
                  DEFAULT_POLICY_ID
              )
              self.preprocessor = (
                  self._ioctx.worker.preprocessors.get(
                      DEFAULT_POLICY_ID
                  )
                  if not self._ioctx.config.get(
                      "_disable_preprocessors", False
                  )
                  else None
              )
          print(
              f"DatasetReader {self._ioctx.worker_index} has {ds.count()}, samples."
          )
  
          def iterator():
              while True:
                  ds = self._dataset.random_shuffle(seed=seed)
                  yield from ds.iter_rows()
  
          self._iter = iterator()
      else:
          self._iter = None
  
imports: import logging
  import math
  from pathlib import Path
  import re
  import numpy
  from typing import List, Tuple, TYPE_CHECKING, Optional
  import zipfile
  import ray.data
  from ray.rllib.offline.input_reader import InputReader
  from ray.rllib.offline.io_context import IOContext
  from ray.rllib.offline.json_reader import from_json_data, postprocess_actions
  from ray.rllib.policy.sample_batch import concat_samples, SampleBatch, DEFAULT_POLICY_ID
  from ray.rllib.utils.annotations import override, PublicAPI
  from ray.rllib.utils.typing import SampleBatchType
label_desc: Use image node_id 1 to create a new DatasetReader object from inherited base classes: InputReader with example: obj = DatasetReader(ds, ioctx)