code: def test_apply_batch_with_type(self):
      pdf = self.pdf
      psdf = ps.from_pandas(pdf)
  
      def identify1(x) -> ps.DataFrame[int, int]:
          return x
  
      # Type hints set the default column names, and we use default index for
      # pandas API on Spark. Here we ignore both diff.
      actual = psdf.pandas_on_spark.apply_batch(identify1)
      expected = pdf
      self.assert_eq(
          sorted(actual["c0"].to_numpy()),
          sorted(expected["a"].to_numpy()),
      )
      self.assert_eq(
          sorted(actual["c1"].to_numpy()),
          sorted(expected["b"].to_numpy()),
      )
  
      def identify2(
          x,
      ) -> ps.DataFrame[slice("a", int), slice("b", int)]:  # noqa: F405
          return x
  
      actual = psdf.pandas_on_spark.apply_batch(identify2)
      expected = pdf
      self.assert_eq(
          sorted(actual["a"].to_numpy()),
          sorted(expected["a"].to_numpy()),
      )
      self.assert_eq(
          sorted(actual["b"].to_numpy()),
          sorted(expected["b"].to_numpy()),
      )
  
      pdf = pd.DataFrame(
          {
              "a": [1, 2, 3, 4, 5, 6, 7, 8, 9],
              "b": [[e] for e in [4, 5, 6, 3, 2, 1, 0, 0, 0]],
          },
          index=np.random.rand(9),
      )
      psdf = ps.from_pandas(pdf)
  
      def identify3(x) -> ps.DataFrame[float, [int, List[int]]]:
          return x
  
      actual = psdf.pandas_on_spark.apply_batch(identify3)
      actual.columns = ["a", "b"]
      self.assert_eq(actual, pdf)
  
      # For NumPy typing, NumPy version should be 1.21+
      if LooseVersion(np.__version__) >= LooseVersion("1.21"):
          import numpy.typing as ntp
  
          psdf = ps.from_pandas(pdf)
  
          def identify4(
              x,
          ) -> ps.DataFrame[float, [int, ntp.NDArray[int]]]:
              return x
  
          actual = psdf.pandas_on_spark.apply_batch(identify4)
          actual.columns = ["a", "b"]
          self.assert_eq(actual, pdf)
  
      arrays = [
          [1, 2, 3, 4, 5, 6, 7, 8, 9],
          ["a", "b", "c", "d", "e", "f", "g", "h", "i"],
      ]
      idx = pd.MultiIndex.from_arrays(arrays, names=("number", "color"))
      pdf = pd.DataFrame(
          {
              "a": [1, 2, 3, 4, 5, 6, 7, 8, 9],
              "b": [[e] for e in [4, 5, 6, 3, 2, 1, 0, 0, 0]],
          },
          index=idx,
      )
      psdf = ps.from_pandas(pdf)
  
      def identify4(x) -> ps.DataFrame[[int, str], [int, List[int]]]:
          return x
  
      actual = psdf.pandas_on_spark.apply_batch(identify4)
      actual.index.names = ["number", "color"]
      actual.columns = ["a", "b"]
      self.assert_eq(actual, pdf)
  
      def identify5(
          x,
      ) -> ps.DataFrame[
          [("number", int), ("color", str)],
          [("a", int), ("b", List[int])],  # noqa: F405
      ]:
          return x
  
      actual = psdf.pandas_on_spark.apply_batch(identify5)
      self.assert_eq(actual, pdf)
  
imports: from datetime import datetime
  import sys
  import unittest
  from typing import List
  import numpy
  import pandas
  from pyspark import pandas
  from pyspark.loose_version import LooseVersion
  from pyspark.pandas.config import option_context
  from pyspark.testing.pandasutils import ComparisonTestBase
  from pyspark.testing.sqlutils import SQLTestUtils
label_desc: Use image node_id 5 for calling the FrameApplyFunctionMixin obj's underlying member method code with example usage: obj.test_apply_batch_with_type() and returns: x, x, x, x, x, x