code: def format_batches(
      block_iter: Iterator[Batch],
      batch_format: Optional[str],
      stats: Optional[DatasetStats] = None,
  ) -> Iterator[Batch]:
      """Given an iterator of blocks, returns an iterator of formatted batches.
  
      Args:
          block_iter: An iterator over blocks.
          batch_format: The batch format to use.
          stats: An optional stats object to record formatting times.
  
      Returns:
          An iterator over batch index and the formatted batch.
      """
      for batch in block_iter:
          with stats.iter_format_batch_s.timer() if stats else nullcontext():
              formatted_batch = BlockAccessor.for_block(
                  batch.data
              ).to_batch_format(batch_format)
          yield Batch(batch.batch_idx, formatted_batch)
  
imports: import logging
  import threading
  from contextlib import nullcontext
  from typing import Any, Callable, Iterator, List, Optional, Tuple
  import ray
  from ray.actor import ActorHandle
  from ray.data._internal.batcher import Batcher, ShufflingBatcher
  from ray.data._internal.block_batching.interfaces import Batch, BlockPrefetcher, CollatedBatch
  from ray.data._internal.stats import DatasetStats
  from ray.data.block import Block, BlockAccessor, DataBatch
  from ray.types import ObjectRef
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 4 for calling a global function with example usage: format_batches(block_iter, batch_format, stats) without return types