[Pandas]groupby 동작에 대해서 (2편)

안녕하세요. Teus입니다.

지난 포스팅을 통해서 Pandas DataFrame을 Groupby 할 경우, 어떻게 Groupby된 Object를 반환하는지 확인 하였습니다.
groupby 이후에 일반적으로 groupby.mean() or .median()과 같은 method Chain을 많이 사용합니다.
이번 포스팅에서는 이 Method Chain이 어떻게 동작하는지 간단히만 살펴보도록 하겠습니다.
(pandas.DataFrame.groupby().mean() 기준으로 살펴봅니다)

---

출처 : pandas/core/groupby/groupby.py#L2081

    @final
    @Substitution(name="groupby")
    @Substitution(see_also=_common_see_also)
    def mean(
        self,
        #lib.no_default는 false랑 다르게, cython에서 default값이 없다고 알려주는 용도로 보임
        numeric_only: bool | lib.NoDefault = lib.no_default,
        engine: str = "cython",
        engine_kwargs: dict[str, bool] | None = None,
    ):        
        #DataFrame의 Data가 Numeric인지 여부를 사전에 체크
        #numeric_only가 일반적으로 lib.no_default로 들어오기 때문에
        #bool로된 값을 별도로 가져감
        numeric_only_bool = self._resolve_numeric_only("mean", numeric_only, axis=0)

        if maybe_use_numba(engine):
            from pandas.core._numba.kernels import sliding_mean

            return self._numba_agg_general(sliding_mean, engine_kwargs)
        else:
            result = self._cython_agg_general(
                "mean",
                alt=lambda x: Series(x).mean(numeric_only=numeric_only_bool),
                numeric_only=numeric_only,
            )
            return result.__finalize__(self.obj, method="groupby")

mean() method는 Numeric 함수인 만큼, 진행 전에 DataFrame의 데이터들이 Numeric인지 여부를 체크합니다.
특이사항으로, lib.no_defualt라는 값이 사용되는데, 이 값은 cython을 위해서 별도로 준비된 값으로 보시면 될것 같습니다.
numba컴파일러는 사용할 것이 아니므로, _cython_agg_general에 대해서 좀더 보도록 하겠습니다.

출처 : pandas/core/groupby/groupby.py#L1743

    @final
    def _cython_agg_general(
        self,
        how: str,
        alt: Callable,
        numeric_only: bool | lib.NoDefault,
        min_count: int = -1,
        ignore_failures: bool = True,
        **kwargs,
    ):
        numeric_only_bool = self._resolve_numeric_only(how, numeric_only, axis=0)

        #type of data = Manager2D
        data = self._get_data_to_aggregate()
        is_ser = data.ndim == 1

        orig_len = len(data)
        if numeric_only_bool:
            if is_ser and not is_numeric_dtype(self._selected_obj.dtype):
                #input 이상에 따른 예외처리 Code
            elif not is_ser:
                data = data.get_numeric_data(copy=False)

        def array_func(values: ArrayLike) -> ArrayLike:
            try:
                result = self.grouper._cython_operation(
                    "aggregate",
                    values,
                    how,
                    axis=data.ndim - 1,
                    min_count=min_count,
                    **kwargs,
                )
            except NotImplementedError:
                result = self._agg_py_fallback(values, ndim=data.ndim, alt=alt)

            return result

        new_mgr = data.grouped_reduce(array_func, ignore_failures=ignore_failures)

        if not is_ser and len(new_mgr) < orig_len:
            warn_dropping_nuisance_columns_deprecated(type(self), how, numeric_only)

        res = self._wrap_agged_manager(new_mgr)
        if is_ser:
            res.index = self.grouper.result_index
            return self._reindex_output(res)
        else:
            return res

이 단계에서는 manager2D type의 Data를 반환하고,
manger2D 의 부모클래스인 BaseBlockManager가 가지고있는 group_reduce method를 사용하는데,
이때 array_func를 집어넣어주어서 array_func가 적용된 새로운 manager 객체를 만드는것을 추측해볼 수 있습니다.

출처 : pandas/core/internals/managers.py#L1472

    def grouped_reduce(self: T, func: Callable, ignore_failures: bool = False) -> T:  
        result_blocks: list[Block] = []
        dropped_any = False

        for blk in self.blocks:
            # blk's type is object
            if blk.is_object:
                # object 값이 있을경우 mean()에서 에러가 발생 하므로
                # column 단위로 나눠서 apply를 진행
                # -> object type이나, 실제 값은 float or int인 경우
                for sb in blk._split():
                    try:
                        applied = sb.apply(func)
                    except (TypeError, NotImplementedError):
                        if not ignore_failures:
                            raise
                        dropped_any = True
                        continue
                    result_blocks = extend_blocks(applied, result_blocks)
            # not type of object
            else:
                # object type이 아닌 경우
                # 이경우 apply(mean)이 실패하면 전체 불가능이므로 예외처리
                # 아닌경우 float / int이므로 정상 처리
                try:
                    applied = blk.apply(func)
                except (TypeError, NotImplementedError):
                    if not ignore_failures:
                        raise
                    dropped_any = True
                    continue
                result_blocks = extend_blocks(applied, result_blocks)

        if len(result_blocks) == 0:
            index = Index([None])  # placeholder
        else:
            index = Index(range(result_blocks[0].values.shape[-1]))

        if dropped_any:
            # faster to skip _combine if we haven't dropped any blocks
            return self._combine(result_blocks, copy=False, index=index)

        return type(self).from_blocks(result_blocks, [self.axes[0], index])

출처 : pandas/core/internals/blocks.py#L398

    @final
    def _split(self) -> list[Block]:
        """
        self가 2차원 block일 때 col단위로 쪼개서 list형태로 반환
        """
        assert self.ndim == 2

        new_blocks = []
        for i, ref_loc in enumerate(self._mgr_locs):
            vals = self.values[slice(i, i + 1)]

            bp = BlockPlacement(ref_loc)
            nb = type(self)(vals, placement=bp, ndim=2)
            new_blocks.append(nb)
        return new_blocks

def extend_blocks(result, blocks=None) -> list[Block]:
    """return a new extended blocks, given the result"""
    if blocks is None:
        blocks = []
    if isinstance(result, list):
        for r in result:
            if isinstance(r, list):
                blocks.extend(r)
            else:
                blocks.append(r)
    else:
        assert isinstance(result, Block), type(result)
        blocks.append(result)
    return blocks

grouped_reduce같은 경우, isnumeric 여부에 따라서 col단위 혹은 DataFrame전체 대상으로 apply(func)를 진행하는 것을 알 수 있습니다.
그 결과 result_blocks 안에는 [funced_block1...]의 형태로 존재할 것이고, 이 result_blocks를 concat해준 뒤
새로운 block으로 만들어서 반환해주는 것을 알 수 있습니다.

---

👀결국 정리해보면

1. groupby object를 mean() method 사용
2. how 가 mean이나 med같이 사전에 정의되어있으면 cython 함수사용
3. pandas DataFrame의 내부의 block단위로 함수 적용
4. Block의 결과를 모아서 새로운 DataFrame생성
5. 반환 하고 함수실행끝!

이번 포스팅을 통해 알 수 있는 사실은, Pandas DataFrame은 Low Level에서 block이라는 단위로 Data를 처리하는것을 알 수 있습니다.
추후에 기회가 된다면 해당 부분에 대해서도 알아보는 시간을 가져볼까 합니다.

이상입니다! 즐거운하루 되세요