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,
          }
      }
code: def next(self) -> SampleBatchType:
      # next() should not get called on None DatasetReader.
      assert self._iter is not None
      ret = []
      count = 0
      while count < self.batch_size:
          d = next(self._iter)
          # Columns like obs are compressed when written by DatasetWriter.
          d = from_json_data(d, self._ioctx.worker)
          count += d.count
          d = self._preprocess_if_needed(d)
          d = postprocess_actions(d, self._ioctx)
          d = self._postprocess_if_needed(d)
          ret.append(d)
      ret = concat_samples(ret)
      return ret
  
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 2 for calling the DatasetReader obj's underlying member method code with example usage: obj.next() and returns: ret