def apply(self, func: Callable, axis: Axis=0, args: Sequence[Any]=(), **kwds: Any) -> Union[('Series', 'DataFrame', 'Index')]:
    '\n        Apply a function along an axis of the DataFrame.\n\n        Objects passed to the function are Series objects whose index is\n        either the DataFrame\'s index (``axis=0``) or the DataFrame\'s columns\n        (``axis=1``).\n\n        See also `Transform and apply a function\n        <https://koalas.readthedocs.io/en/latest/user_guide/transform_apply.html>`_.\n\n        .. note:: when `axis` is 0 or \'index\', the `func` is unable to access\n            to the whole input series. pandas-on-Spark internally splits the input series into\n            multiple batches and calls `func` with each batch multiple times. Therefore, operations\n            such as global aggregations are impossible. See the example below.\n\n            >>> # This case does not return the length of whole series but of the batch internally\n            ... # used.\n            ... def length(s) -> int:\n            ...     return len(s)\n            ...\n            >>> df = ps.DataFrame({\'A\': range(1000)})\n            >>> df.apply(length, axis=0)  # doctest: +SKIP\n            0     83\n            1     83\n            2     83\n            ...\n            10    83\n            11    83\n            dtype: int32\n\n        .. note:: this API executes the function once to infer the type which is\n            potentially expensive, for instance, when the dataset is created after\n            aggregations or sorting.\n\n            To avoid this, specify the return type as `Series` or scalar value in ``func``,\n            for instance, as below:\n\n            >>> def square(s) -> ps.Series[np.int32]:\n            ...     return s ** 2\n\n            pandas-on-Spark uses return type hint and does not try to infer the type.\n\n            In case when axis is 1, it requires to specify `DataFrame` or scalar value\n            with type hints as below:\n\n            >>> def plus_one(x) -> ps.DataFrame[int, [float, float]]:\n            ...     return x + 1\n\n            If the return type is specified as `DataFrame`, the output column names become\n            `c0, c1, c2 ... cn`. These names are positionally mapped to the returned\n            DataFrame in ``func``.\n\n            To specify the column names, you can assign them in a pandas friendly style as below:\n\n            >>> def plus_one(x) -> ps.DataFrame[("index", int), [("a", float), ("b", float)]]:\n            ...     return x + 1\n\n            >>> pdf = pd.DataFrame({\'a\': [1, 2, 3], \'b\': [3, 4, 5]})\n            >>> def plus_one(x) -> ps.DataFrame[\n            ...         (pdf.index.name, pdf.index.dtype), zip(pdf.dtypes, pdf.columns)]:\n            ...     return x + 1\n\n        Parameters\n        ----------\n        func : function\n            Function to apply to each column or row.\n        axis : {0 or \'index\', 1 or \'columns\'}, default 0\n            Axis along which the function is applied:\n\n            * 0 or \'index\': apply function to each column.\n            * 1 or \'columns\': apply function to each row.\n        args : tuple\n            Positional arguments to pass to `func` in addition to the\n            array/series.\n        **kwds\n            Additional keyword arguments to pass as keywords arguments to\n            `func`.\n\n        Returns\n        -------\n        Series or DataFrame\n            Result of applying ``func`` along the given axis of the\n            DataFrame.\n\n        See Also\n        --------\n        DataFrame.applymap : For elementwise operations.\n        DataFrame.aggregate : Only perform aggregating type operations.\n        DataFrame.transform : Only perform transforming type operations.\n        Series.apply : The equivalent function for Series.\n\n        Examples\n        --------\n        >>> df = ps.DataFrame([[4, 9]] * 3, columns=[\'A\', \'B\'])\n        >>> df\n           A  B\n        0  4  9\n        1  4  9\n        2  4  9\n\n        Using a numpy universal function (in this case the same as\n        ``np.sqrt(df)``):\n\n        >>> def sqrt(x) -> ps.Series[float]:\n        ...     return np.sqrt(x)\n        ...\n        >>> df.apply(sqrt, axis=0)\n             A    B\n        0  2.0  3.0\n        1  2.0  3.0\n        2  2.0  3.0\n\n        You can omit the type hint and let pandas-on-Spark infer its type.\n\n        >>> df.apply(np.sqrt, axis=0)\n             A    B\n        0  2.0  3.0\n        1  2.0  3.0\n        2  2.0  3.0\n\n        When `axis` is 1 or \'columns\', it applies the function for each row.\n\n        >>> def summation(x) -> np.int64:\n        ...     return np.sum(x)\n        ...\n        >>> df.apply(summation, axis=1)\n        0    13\n        1    13\n        2    13\n        dtype: int64\n\n        Likewise, you can omit the type hint and let pandas-on-Spark infer its type.\n\n        >>> df.apply(np.sum, axis=1)\n        0    13\n        1    13\n        2    13\n        dtype: int64\n\n        >>> df.apply(max, axis=1)\n        0    9\n        1    9\n        2    9\n        dtype: int64\n\n        Returning a list-like will result in a Series\n\n        >>> df.apply(lambda x: [1, 2], axis=1)\n        0    [1, 2]\n        1    [1, 2]\n        2    [1, 2]\n        dtype: object\n\n        In order to specify the types when `axis` is \'1\', it should use DataFrame[...]\n        annotation. In this case, the column names are automatically generated.\n\n        >>> def identify(x) -> ps.DataFrame[(\'index\', int), [(\'A\', np.int64), (\'B\', np.int64)]]:\n        ...     return x\n        ...\n        >>> df.apply(identify, axis=1)  # doctest: +NORMALIZE_WHITESPACE\n               A  B\n        index\n        0      4  9\n        1      4  9\n        2      4  9\n\n        You can also specify extra arguments.\n\n        >>> def plus_two(a, b, c) -> ps.DataFrame[np.int64, [np.int64, np.int64]]:\n        ...     return a + b + c\n        ...\n        >>> df.apply(plus_two, axis=1, args=(1,), c=3)\n           c0  c1\n        0   8  13\n        1   8  13\n        2   8  13\n        '
    from pyspark.pandas.groupby import GroupBy
    from pyspark.pandas.series import first_series
    if (not isinstance(func, types.FunctionType)):
        assert callable(func), 'the first argument should be a callable function.'
        f = func
        func = (lambda *args, **kwargs: f(*args, **kwargs))
    axis = validate_axis(axis)
    should_return_series = False
    spec = inspect.getfullargspec(func)
    return_sig = spec.annotations.get('return', None)
    should_infer_schema = (return_sig is None)
    should_retain_index = should_infer_schema

    def apply_func(pdf: pd.DataFrame) -> pd.DataFrame:
        pdf_or_pser = pdf.apply(func, axis=axis, args=args, **kwds)
        if isinstance(pdf_or_pser, pd.Series):
            return pdf_or_pser.to_frame()
        else:
            return pdf_or_pser
    self_applied: DataFrame = DataFrame(self._internal.resolved_copy)
    column_labels: Optional[List[Label]] = None
    if should_infer_schema:
        log_advice('If the type hints is not specified for `apply`, it is expensive to infer the data type internally.')
        limit = get_option('compute.shortcut_limit')
        pdf = self_applied.head((limit + 1))._to_internal_pandas()
        applied = pdf.apply(func, axis=axis, args=args, **kwds)
        psser_or_psdf = ps.from_pandas(applied)
        if (len(pdf) <= limit):
            return psser_or_psdf
        psdf = psser_or_psdf
        if isinstance(psser_or_psdf, ps.Series):
            should_return_series = True
            psdf = psser_or_psdf._psdf
        index_fields = [field.normalize_spark_type() for field in psdf._internal.index_fields]
        data_fields = [field.normalize_spark_type() for field in psdf._internal.data_fields]
        return_schema = StructType([field.struct_field for field in (index_fields + data_fields)])
        output_func = GroupBy._make_pandas_df_builder_func(self_applied, apply_func, return_schema, retain_index=should_retain_index)
        sdf = self_applied._internal.to_internal_spark_frame.mapInPandas((lambda iterator: map(output_func, iterator)), schema=return_schema)
        internal = psdf._internal.with_new_sdf(spark_frame=sdf, index_fields=index_fields, data_fields=data_fields)
    else:
        return_type = infer_return_type(func)
        require_index_axis = isinstance(return_type, SeriesType)
        require_column_axis = isinstance(return_type, DataFrameType)
        index_fields = None
        if require_index_axis:
            if (axis != 0):
                raise TypeError(("The given function should specify a scalar or a series as its type hints when axis is 0 or 'index'; however, the return type was %s" % return_sig))
            dtype = cast(SeriesType, return_type).dtype
            spark_type = cast(SeriesType, return_type).spark_type
            data_fields = [InternalField(dtype=dtype, struct_field=StructField(name=name, dataType=spark_type)) for name in self_applied.columns]
            return_schema = StructType([field.struct_field for field in data_fields])
        elif require_column_axis:
            if (axis != 1):
                raise TypeError(("The given function should specify a scalar or a frame as its type hints when axis is 1 or 'column'; however, the return type was %s" % return_sig))
            index_fields = cast(DataFrameType, return_type).index_fields
            should_retain_index = (len(index_fields) > 0)
            data_fields = cast(DataFrameType, return_type).data_fields
            return_schema = cast(DataFrameType, return_type).spark_type
        else:
            should_return_series = True
            spark_type = cast(ScalarType, return_type).spark_type
            dtype = cast(ScalarType, return_type).dtype
            data_fields = [InternalField(dtype=dtype, struct_field=StructField(name=SPARK_DEFAULT_SERIES_NAME, dataType=spark_type))]
            return_schema = StructType([field.struct_field for field in data_fields])
            column_labels = [None]
        output_func = GroupBy._make_pandas_df_builder_func(self_applied, apply_func, return_schema, retain_index=should_retain_index)
        sdf = self_applied._internal.to_internal_spark_frame.mapInPandas((lambda iterator: map(output_func, iterator)), schema=return_schema)
        index_spark_columns = None
        index_names: Optional[List[Optional[Tuple[(Any, ...)]]]] = None
        if should_retain_index:
            index_spark_columns = [scol_for(sdf, index_field.struct_field.name) for index_field in index_fields]
            if (not any([SPARK_INDEX_NAME_PATTERN.match(index_field.struct_field.name) for index_field in index_fields])):
                index_names = [(index_field.struct_field.name,) for index_field in index_fields]
        internal = InternalFrame(spark_frame=sdf, index_names=index_names, index_spark_columns=index_spark_columns, index_fields=index_fields, data_fields=data_fields, column_labels=column_labels)
    result: DataFrame = DataFrame(internal)
    if should_return_series:
        return first_series(result)
    else:
        return result
Apply a function along an axis of the DataFrame.

Objects passed to the function are Series objects whose index is
either the DataFrame's index (``axis=0``) or the DataFrame's columns
(``axis=1``).

See also `Transform and apply a function
<https://koalas.readthedocs.io/en/latest/user_guide/transform_apply.html>`_.

.. note:: when `axis` is 0 or 'index', the `func` is unable to access
    to the whole input series. pandas-on-Spark internally splits the input series into
    multiple batches and calls `func` with each batch multiple times. Therefore, operations
    such as global aggregations are impossible. See the example below.

    >>> # This case does not return the length of whole series but of the batch internally
    ... # used.
    ... def length(s) -> int:
    ...     return len(s)
    ...
    >>> df = ps.DataFrame({'A': range(1000)})
    >>> df.apply(length, axis=0)  # doctest: +SKIP
    0     83
    1     83
    2     83
    ...
    10    83
    11    83
    dtype: int32

.. note:: this API executes the function once to infer the type which is
    potentially expensive, for instance, when the dataset is created after
    aggregations or sorting.

    To avoid this, specify the return type as `Series` or scalar value in ``func``,
    for instance, as below:

    >>> def square(s) -> ps.Series[np.int32]:
    ...     return s ** 2

    pandas-on-Spark uses return type hint and does not try to infer the type.

    In case when axis is 1, it requires to specify `DataFrame` or scalar value
    with type hints as below:

    >>> def plus_one(x) -> ps.DataFrame[int, [float, float]]:
    ...     return x + 1

    If the return type is specified as `DataFrame`, the output column names become
    `c0, c1, c2 ... cn`. These names are positionally mapped to the returned
    DataFrame in ``func``.

    To specify the column names, you can assign them in a pandas friendly style as below:

    >>> def plus_one(x) -> ps.DataFrame[("index", int), [("a", float), ("b", float)]]:
    ...     return x + 1

    >>> pdf = pd.DataFrame({'a': [1, 2, 3], 'b': [3, 4, 5]})
    >>> def plus_one(x) -> ps.DataFrame[
    ...         (pdf.index.name, pdf.index.dtype), zip(pdf.dtypes, pdf.columns)]:
    ...     return x + 1

Parameters
----------
func : function
    Function to apply to each column or row.
axis : {0 or 'index', 1 or 'columns'}, default 0
    Axis along which the function is applied:

    * 0 or 'index': apply function to each column.
    * 1 or 'columns': apply function to each row.
args : tuple
    Positional arguments to pass to `func` in addition to the
    array/series.
**kwds
    Additional keyword arguments to pass as keywords arguments to
    `func`.

Returns
-------
Series or DataFrame
    Result of applying ``func`` along the given axis of the
    DataFrame.

See Also
--------
DataFrame.applymap : For elementwise operations.
DataFrame.aggregate : Only perform aggregating type operations.
DataFrame.transform : Only perform transforming type operations.
Series.apply : The equivalent function for Series.

Examples
--------
>>> df = ps.DataFrame([[4, 9]] * 3, columns=['A', 'B'])
>>> df
   A  B
0  4  9
1  4  9
2  4  9

Using a numpy universal function (in this case the same as
``np.sqrt(df)``):

>>> def sqrt(x) -> ps.Series[float]:
...     return np.sqrt(x)
...
>>> df.apply(sqrt, axis=0)
     A    B
0  2.0  3.0
1  2.0  3.0
2  2.0  3.0

You can omit the type hint and let pandas-on-Spark infer its type.

>>> df.apply(np.sqrt, axis=0)
     A    B
0  2.0  3.0
1  2.0  3.0
2  2.0  3.0

When `axis` is 1 or 'columns', it applies the function for each row.

>>> def summation(x) -> np.int64:
...     return np.sum(x)
...
>>> df.apply(summation, axis=1)
0    13
1    13
2    13
dtype: int64

Likewise, you can omit the type hint and let pandas-on-Spark infer its type.

>>> df.apply(np.sum, axis=1)
0    13
1    13
2    13
dtype: int64

>>> df.apply(max, axis=1)
0    9
1    9
2    9
dtype: int64

Returning a list-like will result in a Series

>>> df.apply(lambda x: [1, 2], axis=1)
0    [1, 2]
1    [1, 2]
2    [1, 2]
dtype: object

In order to specify the types when `axis` is '1', it should use DataFrame[...]
annotation. In this case, the column names are automatically generated.

>>> def identify(x) -> ps.DataFrame[('index', int), [('A', np.int64), ('B', np.int64)]]:
...     return x
...
>>> df.apply(identify, axis=1)  # doctest: +NORMALIZE_WHITESPACE
       A  B
index
0      4  9
1      4  9
2      4  9

You can also specify extra arguments.

>>> def plus_two(a, b, c) -> ps.DataFrame[np.int64, [np.int64, np.int64]]:
...     return a + b + c
...
>>> df.apply(plus_two, axis=1, args=(1,), c=3)
   c0  c1
0   8  13
1   8  13
2   8  13
def apply(self, func: Callable, axis: Axis=0, args: Sequence[Any]=(), **kwds: Any) -> Union[('Series', 'DataFrame', 'Index')]:
    '\n        Apply a function along an axis of the DataFrame.\n\n        Objects passed to the function are Series objects whose index is\n        either the DataFrame\'s index (``axis=0``) or the DataFrame\'s columns\n        (``axis=1``).\n\n        See also `Transform and apply a function\n        <https://koalas.readthedocs.io/en/latest/user_guide/transform_apply.html>`_.\n\n        .. note:: when `axis` is 0 or \'index\', the `func` is unable to access\n            to the whole input series. pandas-on-Spark internally splits the input series into\n            multiple batches and calls `func` with each batch multiple times. Therefore, operations\n            such as global aggregations are impossible. See the example below.\n\n            >>> # This case does not return the length of whole series but of the batch internally\n            ... # used.\n            ... def length(s) -> int:\n            ...     return len(s)\n            ...\n            >>> df = ps.DataFrame({\'A\': range(1000)})\n            >>> df.apply(length, axis=0)  # doctest: +SKIP\n            0     83\n            1     83\n            2     83\n            ...\n            10    83\n            11    83\n            dtype: int32\n\n        .. note:: this API executes the function once to infer the type which is\n            potentially expensive, for instance, when the dataset is created after\n            aggregations or sorting.\n\n            To avoid this, specify the return type as `Series` or scalar value in ``func``,\n            for instance, as below:\n\n            >>> def square(s) -> ps.Series[np.int32]:\n            ...     return s ** 2\n\n            pandas-on-Spark uses return type hint and does not try to infer the type.\n\n            In case when axis is 1, it requires to specify `DataFrame` or scalar value\n            with type hints as below:\n\n            >>> def plus_one(x) -> ps.DataFrame[int, [float, float]]:\n            ...     return x + 1\n\n            If the return type is specified as `DataFrame`, the output column names become\n            `c0, c1, c2 ... cn`. These names are positionally mapped to the returned\n            DataFrame in ``func``.\n\n            To specify the column names, you can assign them in a pandas friendly style as below:\n\n            >>> def plus_one(x) -> ps.DataFrame[("index", int), [("a", float), ("b", float)]]:\n            ...     return x + 1\n\n            >>> pdf = pd.DataFrame({\'a\': [1, 2, 3], \'b\': [3, 4, 5]})\n            >>> def plus_one(x) -> ps.DataFrame[\n            ...         (pdf.index.name, pdf.index.dtype), zip(pdf.dtypes, pdf.columns)]:\n            ...     return x + 1\n\n        Parameters\n        ----------\n        func : function\n            Function to apply to each column or row.\n        axis : {0 or \'index\', 1 or \'columns\'}, default 0\n            Axis along which the function is applied:\n\n            * 0 or \'index\': apply function to each column.\n            * 1 or \'columns\': apply function to each row.\n        args : tuple\n            Positional arguments to pass to `func` in addition to the\n            array/series.\n        **kwds\n            Additional keyword arguments to pass as keywords arguments to\n            `func`.\n\n        Returns\n        -------\n        Series or DataFrame\n            Result of applying ``func`` along the given axis of the\n            DataFrame.\n\n        See Also\n        --------\n        DataFrame.applymap : For elementwise operations.\n        DataFrame.aggregate : Only perform aggregating type operations.\n        DataFrame.transform : Only perform transforming type operations.\n        Series.apply : The equivalent function for Series.\n\n        Examples\n        --------\n        >>> df = ps.DataFrame([[4, 9]] * 3, columns=[\'A\', \'B\'])\n        >>> df\n           A  B\n        0  4  9\n        1  4  9\n        2  4  9\n\n        Using a numpy universal function (in this case the same as\n        ``np.sqrt(df)``):\n\n        >>> def sqrt(x) -> ps.Series[float]:\n        ...     return np.sqrt(x)\n        ...\n        >>> df.apply(sqrt, axis=0)\n             A    B\n        0  2.0  3.0\n        1  2.0  3.0\n        2  2.0  3.0\n\n        You can omit the type hint and let pandas-on-Spark infer its type.\n\n        >>> df.apply(np.sqrt, axis=0)\n             A    B\n        0  2.0  3.0\n        1  2.0  3.0\n        2  2.0  3.0\n\n        When `axis` is 1 or \'columns\', it applies the function for each row.\n\n        >>> def summation(x) -> np.int64:\n        ...     return np.sum(x)\n        ...\n        >>> df.apply(summation, axis=1)\n        0    13\n        1    13\n        2    13\n        dtype: int64\n\n        Likewise, you can omit the type hint and let pandas-on-Spark infer its type.\n\n        >>> df.apply(np.sum, axis=1)\n        0    13\n        1    13\n        2    13\n        dtype: int64\n\n        >>> df.apply(max, axis=1)\n        0    9\n        1    9\n        2    9\n        dtype: int64\n\n        Returning a list-like will result in a Series\n\n        >>> df.apply(lambda x: [1, 2], axis=1)\n        0    [1, 2]\n        1    [1, 2]\n        2    [1, 2]\n        dtype: object\n\n        In order to specify the types when `axis` is \'1\', it should use DataFrame[...]\n        annotation. In this case, the column names are automatically generated.\n\n        >>> def identify(x) -> ps.DataFrame[(\'index\', int), [(\'A\', np.int64), (\'B\', np.int64)]]:\n        ...     return x\n        ...\n        >>> df.apply(identify, axis=1)  # doctest: +NORMALIZE_WHITESPACE\n               A  B\n        index\n        0      4  9\n        1      4  9\n        2      4  9\n\n        You can also specify extra arguments.\n\n        >>> def plus_two(a, b, c) -> ps.DataFrame[np.int64, [np.int64, np.int64]]:\n        ...     return a + b + c\n        ...\n        >>> df.apply(plus_two, axis=1, args=(1,), c=3)\n           c0  c1\n        0   8  13\n        1   8  13\n        2   8  13\n        '
    from pyspark.pandas.groupby import GroupBy
    from pyspark.pandas.series import first_series
    if (not isinstance(func, types.FunctionType)):
        assert callable(func), 'the first argument should be a callable function.'
        f = func
        func = (lambda *args, **kwargs: f(*args, **kwargs))
    axis = validate_axis(axis)
    should_return_series = False
    spec = inspect.getfullargspec(func)
    return_sig = spec.annotations.get('return', None)
    should_infer_schema = (return_sig is None)
    should_retain_index = should_infer_schema

    def apply_func(pdf: pd.DataFrame) -> pd.DataFrame:
        pdf_or_pser = pdf.apply(func, axis=axis, args=args, **kwds)
        if isinstance(pdf_or_pser, pd.Series):
            return pdf_or_pser.to_frame()
        else:
            return pdf_or_pser
    self_applied: DataFrame = DataFrame(self._internal.resolved_copy)
    column_labels: Optional[List[Label]] = None
    if should_infer_schema:
        log_advice('If the type hints is not specified for `apply`, it is expensive to infer the data type internally.')
        limit = get_option('compute.shortcut_limit')
        pdf = self_applied.head((limit + 1))._to_internal_pandas()
        applied = pdf.apply(func, axis=axis, args=args, **kwds)
        psser_or_psdf = ps.from_pandas(applied)
        if (len(pdf) <= limit):
            return psser_or_psdf
        psdf = psser_or_psdf
        if isinstance(psser_or_psdf, ps.Series):
            should_return_series = True
            psdf = psser_or_psdf._psdf
        index_fields = [field.normalize_spark_type() for field in psdf._internal.index_fields]
        data_fields = [field.normalize_spark_type() for field in psdf._internal.data_fields]
        return_schema = StructType([field.struct_field for field in (index_fields + data_fields)])
        output_func = GroupBy._make_pandas_df_builder_func(self_applied, apply_func, return_schema, retain_index=should_retain_index)
        sdf = self_applied._internal.to_internal_spark_frame.mapInPandas((lambda iterator: map(output_func, iterator)), schema=return_schema)
        internal = psdf._internal.with_new_sdf(spark_frame=sdf, index_fields=index_fields, data_fields=data_fields)
    else:
        return_type = infer_return_type(func)
        require_index_axis = isinstance(return_type, SeriesType)
        require_column_axis = isinstance(return_type, DataFrameType)
        index_fields = None
        if require_index_axis:
            if (axis != 0):
                raise TypeError(("The given function should specify a scalar or a series as its type hints when axis is 0 or 'index'; however, the return type was %s" % return_sig))
            dtype = cast(SeriesType, return_type).dtype
            spark_type = cast(SeriesType, return_type).spark_type
            data_fields = [InternalField(dtype=dtype, struct_field=StructField(name=name, dataType=spark_type)) for name in self_applied.columns]
            return_schema = StructType([field.struct_field for field in data_fields])
        elif require_column_axis:
            if (axis != 1):
                raise TypeError(("The given function should specify a scalar or a frame as its type hints when axis is 1 or 'column'; however, the return type was %s" % return_sig))
            index_fields = cast(DataFrameType, return_type).index_fields
            should_retain_index = (len(index_fields) > 0)
            data_fields = cast(DataFrameType, return_type).data_fields
            return_schema = cast(DataFrameType, return_type).spark_type
        else:
            should_return_series = True
            spark_type = cast(ScalarType, return_type).spark_type
            dtype = cast(ScalarType, return_type).dtype
            data_fields = [InternalField(dtype=dtype, struct_field=StructField(name=SPARK_DEFAULT_SERIES_NAME, dataType=spark_type))]
            return_schema = StructType([field.struct_field for field in data_fields])
            column_labels = [None]
        output_func = GroupBy._make_pandas_df_builder_func(self_applied, apply_func, return_schema, retain_index=should_retain_index)
        sdf = self_applied._internal.to_internal_spark_frame.mapInPandas((lambda iterator: map(output_func, iterator)), schema=return_schema)
        index_spark_columns = None
        index_names: Optional[List[Optional[Tuple[(Any, ...)]]]] = None
        if should_retain_index:
            index_spark_columns = [scol_for(sdf, index_field.struct_field.name) for index_field in index_fields]
            if (not any([SPARK_INDEX_NAME_PATTERN.match(index_field.struct_field.name) for index_field in index_fields])):
                index_names = [(index_field.struct_field.name,) for index_field in index_fields]
        internal = InternalFrame(spark_frame=sdf, index_names=index_names, index_spark_columns=index_spark_columns, index_fields=index_fields, data_fields=data_fields, column_labels=column_labels)
    result: DataFrame = DataFrame(internal)
    if should_return_series:
        return first_series(result)
    else:
        return result