code: def _create_map_transformer_for_row_based_map_op(
      row_fn: MapTransformCallable[Row, Row],
      init_fn: Optional[Callable[[], None]] = None,
  ) -> MapTransformer:
      """Create a MapTransformer for a row-based map operator
      (e.g. map, flat_map, filter)."""
      transform_fns = [
          # Convert input blocks to rows.
          BlocksToRowsMapTransformFn.instance(),
          # Apply the UDF.
          RowMapTransformFn(row_fn),
          # Convert output rows to blocks.
          BuildOutputBlocksMapTransformFn.for_rows(),
      ]
      return MapTransformer(transform_fns, init_fn=init_fn)
  
imports: import collections
  from types import GeneratorType
  from typing import Any, Callable, Iterable, Iterator, Optional
  import numpy
  import pandas
  import pyarrow
  import ray
  from ray.data._internal.compute import get_compute
  from ray.data._internal.execution.interfaces import PhysicalOperator
  from ray.data._internal.execution.interfaces.task_context import TaskContext
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import BatchMapTransformFn, BlocksToBatchesMapTransformFn, BlocksToRowsMapTransformFn, BuildOutputBlocksMapTransformFn, MapTransformCallable, MapTransformer, Row, RowMapTransformFn
  from ray.data._internal.execution.util import make_callable_class_concurrent
  from ray.data._internal.logical.operators.map_operator import AbstractUDFMap, Filter, FlatMap, MapBatches, MapRows
  from ray.data._internal.numpy_support import is_valid_udf_return
  from ray.data._internal.util import _truncated_repr
  from ray.data.block import Block, BlockAccessor, CallableClass, DataBatch, UserDefinedFunction
  from ray.data.context import DataContext
label_desc: Use image node_id 10 for calling a global function with example usage: _create_map_transformer_for_row_based_map_op(row_fn, init_fn) and returns: MapTransformer