code: def _get_process(
      bento_tag: Tag, api_name: str
  ) -> t.Callable[
      [t.Iterable[RecordBatch]], t.Generator[RecordBatch, None, None]
  ]:
      def process(
          iterator: t.Iterable[RecordBatch],
      ) -> t.Generator[RecordBatch, None, None]:
          svc = _load_bento_spark(bento_tag)
  
          assert (
              api_name in svc.apis
          ), "An error occurred transferring the Bento to the Spark worker."
          inference_api = svc.apis[api_name]
          assert (
              inference_api.func is not None
          ), "Inference API function not defined"
  
          # start bento server
          with reserve_free_port() as port:
              pass
  
          server = serve(bento_tag, port=port)
          Client.wait_until_server_ready("localhost", server.port, 30)
          client = HTTPClient(svc, f"http://localhost:{server.port}")
  
          for batch in iterator:
              func_input = inference_api.input.from_arrow(batch)
              func_output = client.call(api_name, func_input)
              yield inference_api.output.to_arrow(func_output)
  
      return process
  
file_path: BentoML/grpc-client/bentoml/_internal/batch/spark.py
imports: from __future__ import annotations
  import logging
  import os.path
  import tempfile
  import typing
  from typing import TYPE_CHECKING
  from bentoml._internal.utils import reserve_free_port
  from ...bentos import import_bento
  from ...bentos import serve
  from ...client import Client
  from ...client import HTTPClient
  from ...exceptions import BentoMLException
  from ...exceptions import MissingDependencyException
  from ..bento import Bento
  from ..service.loader import load_bento
  from ..tag import Tag
label_desc: Use image node_id 3 for calling a global function with example usage: _get_process(bento_tag, api_name) and returns: process