class_docstr: A MapTransformFn that converts input blocks to batches.
class_docstr_tok: ["A","MapTransformFn","that","converts","input","blocks","to","batches","."]
docstr: Converts input blocks to batches.
docstr_tok: ["Converts","input","blocks","to","batches","."]
code: def __call__(
      self,
      blocks: Iterable[Block],
      _: TaskContext,
  ) -> Iterable[DataBatch]:
      """Converts input blocks to batches."""
      block_iter = iter(blocks)
      first = next(block_iter, None)
      if first is None:
          return []
      blocks = itertools.chain([first], block_iter)
      empty_block = BlockAccessor.for_block(first).builder().build()
      # Don't hold the first block in memory, so we reset the reference.
      first = None
  
      # Ensure that zero-copy batch views are copied so mutating UDFs don't error.
      formatted_batch_iter = batch_blocks(
          blocks=blocks,
          stats=None,
          batch_size=self._batch_size,
          batch_format=self._batch_format,
          ensure_copy=self._ensure_copy,
      )
  
      first = next(formatted_batch_iter, None)
      if first is None:
          # If the input blocks are all empty, then yield an empty block with same
          # format as the input blocks.
          return [empty_block]
      else:
          return itertools.chain([first], formatted_batch_iter)
  
imports: import itertools
  from abc import abstractmethod
  from enum import Enum
  from typing import Any, Callable, Dict, Iterable, List, Optional, TypeVar, Union
  from ray.data._internal.block_batching.block_batching import batch_blocks
  from ray.data._internal.execution.interfaces.task_context import TaskContext
  from ray.data._internal.output_buffer import BlockOutputBuffer
  from ray.data.block import Block, BlockAccessor, DataBatch
label_desc: Use image node_id 2 for calling the BlocksToBatchesMapTransformFn obj's underlying member method code with example usage: obj.__call__(blocks, _) and returns: list, list, itertools