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 _preprocess_if_needed(
      self, batch: SampleBatchType
  ) -> SampleBatchType:
      # TODO: @kourosh, preprocessor is only supported for single agent case.
      if self.preprocessor:
          for key in (SampleBatch.CUR_OBS, SampleBatch.NEXT_OBS):
              if key in batch:
                  batch[key] = np.stack(
                      [
                          self.preprocessor.transform(s)
                          for s in batch[key]
                      ]
                  )
      return batch
  
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 3 for calling the DatasetReader obj's underlying member method code with example usage: obj._preprocess_if_needed(batch) and returns: batch