code: def test_map_batches_generator(ray_start_regular_shared, tmp_path):
      # Set up.
      df = pd.DataFrame({"one": [1, 2, 3], "two": [2, 3, 4]})
      table = pa.Table.from_pandas(df)
      pq.write_table(table, os.path.join(tmp_path, "test1.parquet"))
  
      def pandas_generator(
          batch: pd.DataFrame,
      ) -> Iterator[pd.DataFrame]:
          for i in range(len(batch)):
              yield batch.iloc[[i]] + 1
  
      ds = ray.data.read_parquet(str(tmp_path))
      ds2 = ds.map_batches(
          pandas_generator, batch_size=1, batch_format="pandas"
      )
      ds_list = ds2.take()
      values = sorted([s["one"] for s in ds_list])
      assert values == [2, 3, 4]
      values = sorted([s["two"] for s in ds_list])
      assert values == [3, 4, 5]
  
      def fail_generator(batch):
          for i in range(len(batch)):
              yield i
  
      # Test the wrong return value raises an exception.
      ds = ray.data.read_parquet(str(tmp_path))
      with pytest.raises(ValueError):
          ds_list = ds.map_batches(
              fail_generator, batch_size=2, batch_format="pyarrow"
          ).take()
  
imports: import itertools
  import math
  import os
  import signal
  import threading
  import time
  from typing import Iterator
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import column_udf, column_udf_class, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 12 for calling a global function with example usage: test_map_batches_generator(ray_start_regular_shared, tmp_path) without return types