code: def _parse_op_fn(op: AbstractUDFMap):
      # Note, it's important to define these standalone variables.
      # So the parsed functions won't need to caputure the entire operator, which may not
      # be serializable.
      op_fn = op._fn
      fn_args = op._fn_args or ()
      fn_kwargs = op._fn_kwargs or {}
  
      if isinstance(op._fn, CallableClass):
          fn_constructor_args = op._fn_constructor_args or ()
          fn_constructor_kwargs = op._fn_constructor_kwargs or {}
  
          op_fn = make_callable_class_concurrent(op_fn)
  
          def fn(item: Any) -> Any:
              assert ray.data._cached_fn is not None
              assert ray.data._cached_cls == op_fn
              return ray.data._cached_fn(item, *fn_args, **fn_kwargs)
  
          def init_fn():
              if ray.data._cached_fn is None:
                  ray.data._cached_cls = op_fn
                  ray.data._cached_fn = op_fn(
                      *fn_constructor_args, **fn_constructor_kwargs
                  )
  
      else:
  
          def fn(item: Any) -> Any:
              return op_fn(item, *fn_args, **fn_kwargs)
  
          def init_fn():
              pass
  
      return 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 2 for calling a global function with example usage: _parse_op_fn(op) and returns: fn,  init_fn, ray, op_fn