class_docstr: Represents a grouped dataset created by calling ``Dataset.groupby()``.
  
  The actual groupby is deferred until an aggregation is applied.
class_docstr_tok: ["Represents","a","grouped","dataset","created","by","calling","``","Dataset.groupby","(",")","``",".","The","actual","groupby","is","deferred","until","an","aggregation","is","applied","."]
docstr: Implements an accumulator-based aggregation.
  
  Args:
      aggs: Aggregations to do.
  
  Returns:
      The output is an dataset of ``n + 1`` columns where the first column
      is the groupby key and the second through ``n + 1`` columns are the
      results of the aggregations.
      If groupby key is ``None`` then the key part of return is omitted.
docstr_tok: ["Implements","an","accumulator-based","aggregation",".","Args",":","aggs",":","Aggregations","to","do",".","Returns",":","The","output","is","an","dataset","of","``","n","+","1","``","columns","where","the","first","column","is","the","groupby","key","and","the","second","through","``","n","+","1","``","columns","are","the","results","of","the","aggregations",".","If","groupby","key","is","``","None","``","then","the","key","part","of","return","is","omitted","."]
code: def aggregate(self, *aggs: AggregateFn) -> Dataset:
      """Implements an accumulator-based aggregation.
  
      Args:
          aggs: Aggregations to do.
  
      Returns:
          The output is an dataset of ``n + 1`` columns where the first column
          is the groupby key and the second through ``n + 1`` columns are the
          results of the aggregations.
          If groupby key is ``None`` then the key part of return is omitted.
      """
  
      def do_agg(
          blocks, task_ctx: TaskContext, clear_input_blocks: bool, *_
      ):
          # TODO: implement clear_input_blocks
          stage_info = {}
          if len(aggs) == 0:
              raise ValueError(
                  "Aggregate requires at least one aggregation"
              )
          for agg in aggs:
              agg._validate(self._dataset.schema(fetch_if_missing=True))
          # Handle empty dataset.
          if blocks.initial_num_blocks() == 0:
              return blocks, stage_info
  
          num_mappers = blocks.initial_num_blocks()
          num_reducers = num_mappers
          if self._key is None:
              num_reducers = 1
              boundaries = []
          else:
              boundaries = sort.sample_boundaries(
                  blocks.get_blocks(),
                  SortKey(self._key),
                  num_reducers,
                  task_ctx,
              )
          ctx = DataContext.get_current()
          if ctx.use_push_based_shuffle:
              shuffle_op_cls = PushBasedGroupbyOp
          else:
              shuffle_op_cls = SimpleShuffleGroupbyOp
          shuffle_op = shuffle_op_cls(
              map_args=[boundaries, self._key, aggs],
              reduce_args=[self._key, aggs],
          )
          return shuffle_op.execute(
              blocks,
              num_reducers,
              clear_input_blocks,
              ctx=task_ctx,
          )
  
      plan = self._dataset._plan.with_stage(
          AllToAllStage(
              "Aggregate",
              None,
              do_agg,
              sub_stage_names=[
                  "SortSample",
                  "ShuffleMap",
                  "ShuffleReduce",
              ],
          )
      )
  
      logical_plan = self._dataset._logical_plan
      if logical_plan is not None:
          op = Aggregate(
              logical_plan.dag,
              key=self._key,
              aggs=aggs,
          )
          logical_plan = LogicalPlan(op)
      return Dataset(
          plan,
          logical_plan,
      )
  
imports: from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
  from ._internal.table_block import TableBlockAccessor
  from ray.data._internal import sort
  from ray.data._internal.compute import ComputeStrategy
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.logical.interfaces import LogicalPlan
  from ray.data._internal.logical.operators.all_to_all_operator import Aggregate
  from ray.data._internal.plan import AllToAllStage
  from ray.data._internal.push_based_shuffle import PushBasedShufflePlan
  from ray.data._internal.shuffle import ShuffleOp, SimpleShufflePlan
  from ray.data._internal.sort import SortKey
  from ray.data.aggregate import AggregateFn, Count, Max, Mean, Min, Std, Sum
  from ray.data.aggregate._aggregate import _AggregateOnKeyBase
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata, KeyType, UserDefinedFunction
  from ray.data.context import DataContext
  from ray.data.dataset import DataBatch, Dataset
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 3 for calling the GroupedData obj's underlying member method code with example usage: obj.aggregate() and returns: Dataset, shuffle_op, blocks,  stage_info