code: def __init__(
      self,
      dataset: "MapDataset",
      batch_size: int,
      batch_per_thread: int,
      validation_dataset: Optional["MapDataset"] = None,
      intra_threads: Optional[int] = None,
      inter_threads: Optional[int] = None,
  ) -> None:
      node_num, core_num = get_node_and_core_number()
  
      self.intra_threads = intra_threads
      self.inter_threads = inter_threads
      if intra_threads is None:
          self.intra_threads = core_num
  
      if inter_threads is None:
          self.inter_threads = 1
  
      if batch_size > 0:
          num_parts = dataset.xshards.num_partitions()
          if num_parts != node_num:
              dataset.xshards = dataset.xshards.repartition(node_num)
          invalidInputError(
              batch_size % node_num == 0,
              "batch_size should be a multiple of num_shards, got"
              " batch_size {}, node_num {}".format(
                  batch_size, node_num
              ),
          )
          batch_per_shard = batch_size // node_num
          self.drop_remainder = True
      elif batch_per_thread > 0:
          batch_per_shard = batch_per_thread
          self.drop_remainder = False
      else:
          invalidInputError(
              False,
              "one of batch_size or batch_per_thread must be larger than 0",
          )
  
      self.rdd = dataset.as_graph_rdd(
          batch_per_shard, drop_remainder=self.drop_remainder
      ).cache()
      meta_info = self.rdd.map(lambda x: x[1]).first()
      tensor_structure = meta_info["tensor_structure"]
      self.init_op_name = meta_info["init_op_name"]
      self.output_names = meta_info["output_names"]
      self.output_types = meta_info["output_types"]
      self.table_init_op = meta_info["table_init_op"]
  
      if validation_dataset is not None:
          self.val_rdd = validation_dataset.as_graph_rdd(
              batch_per_shard, False
          ).cache()
          meta_info = self.val_rdd.map(lambda x: x[1]).first()
          self.val_init_op_name = meta_info["init_op_name"]
          self.val_output_names = meta_info["output_names"]
          self.val_output_types = meta_info["output_types"]
      else:
          self.val_rdd = None  # type: ignore
          self.val_init_op_name = None
          self.val_output_names = None
          self.val_output_types = None
  
      super().__init__(
          tensor_structure,
          batch_size=batch_size,
          batch_per_thread=batch_per_thread,
          hard_code_batch_size=False,
      )
      self.shard_index_op_name = None
      self.validation_dataset = validation_dataset
  
file_path: BigDL/python/orca/src/bigdl/orca/data/tf/tf1_data.py
imports: from bigdl.dllib.utils.common import get_node_and_core_number
  from bigdl.dllib.utils.file_utils import callZooFunc
  from bigdl.dllib.feature.common import FeatureSet
  from bigdl.orca.tfpark import TFDataset
  from bigdl.dllib.utils.log4Error import invalidInputError
  from typing import TYPE_CHECKING, Optional, Any
  import logging
label_desc: Use image node_id 1 to create a new TF1Dataset object from inherited base classes: TFDataset with example: obj = TF1Dataset(dataset, batch_size, batch_per_thread, validation_dataset, intra_threads, inter_threads)