DEVOCEAN

import pandas as pd
temp_df = pd.DataFrame({"a" : [i*1 for i in range(100)],
                        "b" : [i*2 for i in range(100)],
                        "c" : [i*3 for i in range(100)],
                        "d" : [i*4 for i in range(100)],
                        "e" : [i*5 for i in range(100)]})
#"a" col에 ElementWise하게 제곱을 적용
temp_df["a"].apply(lambda x : x*x)

class DataFrame(NDFrame, OpsMixin):
    ...
    def apply(
        self,
        func: AggFuncType,
        axis: Axis = 0,
        raw: bool = False,
        result_type=None,
        args=(),
        **kwargs,
    ):

        from pandas.core.apply import frame_apply

        op = frame_apply(
            self,
            func=func,
            axis=axis,
            raw=raw,
            result_type=result_type,
            args=args,
            kwargs=kwargs,
        )
        return op.apply().__finalize__(self, method="apply")
    

def frame_apply(
    obj: DataFrame,
    func: AggFuncType,
    axis: Axis = 0,
    raw: bool = False,
    result_type: str | None = None,
    args=None,
    kwargs=None,
) -> FrameApply:
    """construct and return a row or column based frame apply object"""
    axis = obj._get_axis_number(axis)
    klass: type[FrameApply]
    if axis == 0:
        klass = FrameRowApply
    elif axis == 1:
        klass = FrameColumnApply

    return klass(
        obj,
        func,
        raw=raw,
        result_type=result_type,
        args=args,
        kwargs=kwargs,
    )

class FrameApply(NDFrameApply):
    obj: DataFrame
    ...
    def apply(self) -> DataFrame | Series:
    """compute the results"""
    # dispatch to agg
    if is_list_like(self.f):
        return self.apply_multiple()

    # all empty
    if len(self.columns) == 0 and len(self.index) == 0:
        return self.apply_empty_result()

    # string dispatch
    if isinstance(self.f, str):
        return self.apply_str()

    # ufunc
    elif isinstance(self.f, np.ufunc):
        with np.errstate(all="ignore"):
            results = self.obj._mgr.apply("apply", func=self.f)
        # _constructor will retain self.index and self.columns
        return self.obj._constructor(data=results)

    # broadcasting
    if self.result_type == "broadcast":
        return self.apply_broadcast(self.obj)

    # one axis empty
    elif not all(self.obj.shape):
        return self.apply_empty_result()

    # raw
    elif self.raw:
        return self.apply_raw()

    return self.apply_standard()

class Apply(metaclass=abc.ABCMeta):
    axis: int

    def __init__(
        self,
        obj: AggObjType,
        func,
        raw: bool,
        result_type: str | None,
        args,
        kwargs,
    ):
        self.obj = obj
        self.raw = raw
        self.args = args or ()
        self.kwargs = kwargs or {}

        if result_type not in [None, "reduce", "broadcast", "expand"]:
            raise ValueError(
                "invalid value for result_type, must be one "
                "of {None, 'reduce', 'broadcast', 'expand'}"
            )

        self.result_type = result_type
        
        # Funciton을 적용하기 위해서 함수를 저장
        # curry if needed
        if (
            (kwargs or args)
            and not isinstance(func, (np.ufunc, str))
            and not is_list_like(func)
        ):

            def f(x):
                return func(x, *args, **kwargs)

        else:
            f = func

        self.orig_f: AggFuncType = func
        self.f: AggFuncType = f

    @property
    def series_generator(self):
        return (self.obj._ixs(i, axis=1) for i in range(len(self.columns)))
        

    def apply_standard(self):
        results, res_index = self.apply_series_generator()

        # wrap results
        return self.wrap_results(results, res_index)

    def apply_series_generator(self) -> tuple[ResType, Index]:
        assert callable(self.f)

        series_gen = self.series_generator
        res_index = self.result_index

        results = {}

        with option_context("mode.chained_assignment", None):
            #=====================실제 함수 실행 부분===========================#
            for i, v in enumerate(series_gen):
                # ignore SettingWithCopy here in case the user mutates
                results[i] = self.f(v)
                if isinstance(results[i], ABCSeries):
                    # If we have a view on v, we need to make a copy because
                    #  series_generator will swap out the underlying data
                    results[i] = results[i].copy(deep=False)
            #=====================실제 함수 실행 부분===========================#
        return results, res_index

    def wrap_results(self, results: ResType, res_index: Index) -> DataFrame | Series:
        from pandas import Series

        # see if we can infer the results
        if len(results) > 0 and 0 in results and is_sequence(results[0]):
            return self.wrap_results_for_axis(results, res_index)

        # dict of scalars

        # the default dtype of an empty Series will be `object`, but this
        # code can be hit by df.mean() where the result should have dtype
        # float64 even if it's an empty Series.
        constructor_sliced = self.obj._constructor_sliced
        if constructor_sliced is Series:
            result = create_series_with_explicit_dtype(
                results, dtype_if_empty=np.float64
            )
        else:
            result = constructor_sliced(results)
        result.index = res_index

        return result       

class Series(base.IndexOpsMixin, NDFrame):
    ...
    def apply(
            self,
            func: AggFuncType,
            convert_dtype: bool = True,
            args: tuple[Any, ...] = (),
            **kwargs,
        ) -> DataFrame | Series:
        return SeriesApply(self, func, convert_dtype, args, kwargs).apply()

class SeriesApply(NDFrameApply):
    obj: Series
    axis = 0

    def __init__(
        self,
        obj: Series,
        func: AggFuncType,
        convert_dtype: bool,
        args,
        kwargs,
    ):
        self.convert_dtype = convert_dtype

        super().__init__(
            obj,
            func,
            raw=False,
            result_type=None,
            args=args,
            kwargs=kwargs,
        )

    def apply(self) -> DataFrame | Series:
        obj = self.obj

        if len(obj) == 0:
            return self.apply_empty_result()

        # dispatch to agg
        if is_list_like(self.f):
            return self.apply_multiple()

        if isinstance(self.f, str):
            # if we are a string, try to dispatch
            return self.apply_str()

        return self.apply_standard()
    def apply_standard(self) -> DataFrame | Series:
        f = self.f
        obj = self.obj

        with np.errstate(all="ignore"):
            if isinstance(f, np.ufunc):
                return f(obj)

            # row-wise access
            if is_extension_array_dtype(obj.dtype) and hasattr(obj._values, "map"):
                # GH#23179 some EAs do not have `map`
                mapped = obj._values.map(f)
            else:
                values = obj.astype(object)._values
                # error: Argument 2 to "map_infer" has incompatible type
                # "Union[Callable[..., Any], str, List[Union[Callable[..., Any], str]],
                # Dict[Hashable, Union[Union[Callable[..., Any], str],
                # List[Union[Callable[..., Any], str]]]]]"; expected
                # "Callable[[Any], Any]"
                mapped = lib.map_infer(
                    values,
                    f,  # type: ignore[arg-type]
                    convert=self.convert_dtype,
                )

        if len(mapped) and isinstance(mapped[0], ABCSeries):
            # GH#43986 Need to do list(mapped) in order to get treated as nested
            #  See also GH#25959 regarding EA support
            return obj._constructor_expanddim(list(mapped), index=obj.index)
        else:
            return obj._constructor(mapped, index=obj.index).__finalize__(
                obj, method="apply"
            )

@cython.boundscheck(False)
@cython.wraparound(False)
def map_infer(
    ndarray arr, object f, bint convert=True, bint ignore_na=False
) -> np.ndarray:    
    cdef:
        Py_ssize_t i, n
        ndarray[object] result
        object val

    n = len(arr)
    result = np.empty(n, dtype=object)
    for i in range(n):
        if ignore_na and checknull(arr[i]):
            result[i] = arr[i]
            continue
        val = f(arr[i])

        if cnp.PyArray_IsZeroDim(val):
            # unbox 0-dim arrays, GH#690
            val = val.item()

        result[i] = val

    if convert:
        #maybe_convert_object source : 
        #https://github.com/pandas-dev/pandas/blob/e8093ba372f9adfe79439d90fe74b0b5b6dea9d6/pandas/_libs/lib.pyx#L2395
        return maybe_convert_objects(result,
                                     try_float=False,
                                     convert_datetime=False,
                                     convert_timedelta=False)

    return result

import pandas as pd
import time
row = 5000000
temp_df = pd.DataFrame({"a" : [i*1 for i in range(row)],
                        "b" : [i*2 for i in range(row)],
                        "c" : [i*3 for i in range(row)],
                        "d" : [i*4 for i in range(row)],
                        "e" : [i*5 for i in range(row)]})
def foo(x):    
    return x*x

#======================Row 단위 apply===============================
st = time.time()
temp_df["a"].apply(lambda x : foo(x))
print("Series apply time : ", time.time() - st)

temp_series = temp_df["a"].copy()
st = time.time()
for idx, val in enumerate(temp_series):
    temp_series[idx] = foo(val)
print("Series normal time : ", time.time() - st)

st = time.time()
temp_df["a"] = foo(temp_df["a"])
print("Series Direct Function time : ", time.time() - st)
#======================Row 단위 apply===============================

#======================Col 단위 apply===============================
st = time.time()
temp_df.apply(lambda x : foo(x), axis = 0)
print("DataFrame apply time : ", time.time() - st)


st = time.time()
result = {}
for i in (temp_df._ixs(i, axis=1) for i in range(len(temp_df.columns))):
    result[i.name] = foo(i)
pd.DataFrame(result)
print("DataFrame normal time : ", time.time() - st)

st = time.time()
result = {}
for i in (temp_df._ixs(i, axis=1) for i in range(len(temp_df.columns))):
    result[i.name] = foo(i)
print("DataFrame normal time(without DataFrame Construct) : ", time.time() - st)
#======================Col 단위 apply===============================
"""
Series apply time :  2.672856092453003
Series normal time :  26.50118923187256
Series Direct Function time :  0.09272193908691406
DataFrame apply time :  0.1655576229095459
DataFrame normal time :  0.18553519248962402
DataFrame normal time(without DataFrame Construct) :  0.09574413299560547
"""