code: def _execute_map(cls, ctx, op: "DataFrameGroupByAgg"):
      xdf = cudf if op.gpu else pd
  
      in_data = ctx[op.inputs[0].key]
      if (
          isinstance(in_data, xdf.Series)
          and op.output_types[0] == OutputType.dataframe
      ):
          in_data = cls._series_to_df(in_data, op.gpu)
  
      # map according to map groups
      ret_map_groupbys = dict()
      grouped = cls._get_grouped(op, in_data, ctx)
      grouper = None
      drop_names = False
  
      for input_key, output_key, cols, func in op.pre_funcs:
          if input_key == output_key:
              if (
                  cols is None
                  or getattr(grouped, "_selection", None) is not None
              ):
                  ret_map_groupbys[output_key] = grouped
              else:
                  ret_map_groupbys[output_key] = grouped[cols]
          else:
  
              def _wrapped_func(col):
                  try:
                      return func(col, gpu=op.is_gpu())
                  except TypeError:
                      return col
  
              pre_df = in_data if cols is None else in_data[cols]
              try:
                  pre_df = func(pre_df, gpu=op.is_gpu())
              except TypeError:
                  pre_df = pre_df.transform(_wrapped_func)
  
              if grouper is None:
                  try:
                      grouper = grouped.grouper
                  except (
                      AttributeError
                  ):  # cudf does not have GroupBy.grouper
                      grouper = xdf.Series(
                          grouped.grouping.keys, index=grouped.obj.index
                      )
                      if in_data.ndim == 2:
                          drop_names = True
  
              if drop_names:
                  pre_df = pre_df.drop(
                      columns=grouped.grouping.names, errors="ignore"
                  )
              ret_map_groupbys[output_key] = cls._get_grouped(
                  op, pre_df, ctx, grouper=grouper
              )
  
      agg_dfs = []
      for (
          input_key,
          raw_func_name,
          map_func_name,
          _agg_func_name,
          _custom_reduction,
          _output_key,
          _output_limit,
          kwds,
      ) in op.agg_funcs:
          input_obj = ret_map_groupbys[input_key]
          if map_func_name == "custom_reduction":
              agg_dfs.append(
                  cls._do_custom_agg(raw_func_name, op, in_data)
              )
          else:
              single_func = map_func_name == op.raw_func
              agg_dfs.append(
                  cls._do_predefined_agg(
                      input_obj, map_func_name, single_func, **kwds
                  )
              )
  
      if getattr(op, "size_recorder_name", None) is not None:
          # record_size
          raw_size = estimate_pandas_size(in_data)
          # when agg by a list of methods, agg_size should be sum
          agg_size = sum(
              [estimate_pandas_size(item) for item in agg_dfs]
          )
          size_recorder = ctx.get_remote_object(op.size_recorder_name)
          size_recorder.record(raw_size, agg_size)
  
      ctx[op.outputs[0].key] = tuple(agg_dfs)
  
imports: import functools
  import itertools
  import logging
  import uuid
  from typing import Callable, Dict, List, Union
  import numpy
  import pandas
  from ...None import opcodes
  from ...config import options
  from ...core import ENTITY_TYPE, OutputType
  from ...core.custom_log import redirect_custom_log
  from ...core.context import get_context
  from ...core.operand import OperandStage
  from ...serialization.serializables import Int32Field, Int64Field, AnyField, BoolField, StringField, ListField, DictField
  from ...typing import ChunkType, TileableType
  from ...utils import enter_current_session, lazy_import, pd_release_version, estimate_pandas_size
  from ..arrays import ArrowArray
  from ..core import GROUPBY_TYPE
  from ..merge import DataFrameConcat
  from ..operands import DataFrameOperand, DataFrameOperandMixin, DataFrameShuffleProxy
  from ..reduction.core import ReductionCompiler, ReductionSteps, ReductionAggStep
  from ..reduction.aggregation import is_funcs_aggregate, normalize_reduction_funcs
  from ..utils import parse_index, build_concatenated_rows_frame, is_cudf, concat_on_columns
  from .core import DataFrameGroupByOperand
  from .custom_aggregation import custom_agg_functions
  from .sort import DataFramePSRSGroupbySample, DataFrameGroupbyConcatPivot, DataFrameGroupbySortShuffle
label_desc: Use image node_id 30 for calling the DataFrameGroupByAgg obj's underlying member method code with example usage: obj._execute_map(cls, ctx, op) and returns: func, col