code: def map_batches(
      input_ds: Dataset,
      batch_size: int,
      batch_format: Literal["default", "pandas", "pyarrow", "numpy"],
      concurrency: Optional[Union[int, Tuple[int, int]]] = None,
      num_calls: Optional[int] = 1,
      is_eager_executed: Optional[bool] = False,
  ) -> Dataset:
      assert isinstance(input_ds, MaterializedDataset)
      ds = input_ds
  
      def udf(x):
          return x
  
      class UDFClass:
          def __call__(self, x):
              return x
  
      if concurrency is None:
          fn = udf
      else:
          fn = UDFClass
  
      for _ in range(num_calls):
          ds = ds.map_batches(
              fn,
              batch_format=batch_format,
              batch_size=batch_size,
              concurrency=concurrency,
          )
          if is_eager_executed:
              ds = ds.materialize()
      return ds
  
file_path: ray/release/nightly_tests/dataset/map_batches_benchmark.py
imports: from typing import Optional, Union, Tuple
  import numpy
  import ray
  from ray.data.dataset import Dataset, MaterializedDataset
  from benchmark import Benchmark
  from typing import Literal
label_desc: Use image node_id 1 for calling a global function with example usage: map_batches(input_ds, batch_size, batch_format, concurrency, num_calls, is_eager_executed) and returns: ds, x, x