class_docstr: The implementation for sort-based aggregate tasks.
  
  Aggregate is done in 2 steps: partial aggregate of individual blocks, and
  final aggregate of sorted blocks.
  
  Partial aggregate (`map`): each block is sorted locally, then partitioned into
  smaller blocks according to the boundaries. Each partitioned block is aggregated
  separately, then passed to a final aggregate task.
  
  Final aggregate (`reduce`): each task would receive a block from every worker that
  consists of items in a certain range. It then merges the sorted blocks and
  aggregates on-the-fly.
class_docstr_tok: ["The","implementation","for","sort-based","aggregate","tasks",".","Aggregate","is","done","in","2","steps",":","partial","aggregate","of","individual","blocks",",","and","final","aggregate","of","sorted","blocks",".","Partial","aggregate","(","`","map","`",")",":","each","block","is","sorted","locally",",","then","partitioned","into","smaller","blocks","according","to","the","boundaries",".","Each","partitioned","block","is","aggregated","separately",",","then","passed","to","a","final","aggregate","task",".","Final","aggregate","(","`","reduce","`",")",":","each","task","would","receive","a","block","from","every","worker","that","consists","of","items","in","a","certain","range",".","It","then","merges","the","sorted","blocks","and","aggregates","on-the-fly","."]
code: def __init__(
      self,
      boundaries: List[KeyType],
      key: Optional[str],
      aggs: List[AggregateFn],
  ):
      super().__init__(
          map_args=[boundaries, key, aggs],
          reduce_args=[key, aggs],
      )
  
file_path: ray/python/ray/data/_internal/planner/exchange/aggregate_task_spec.py
imports: from typing import List, Optional, Tuple, Union
  from ray.data._internal.planner.exchange.interfaces import ExchangeTaskSpec
  from ray.data._internal.sort import SortKey
  from ray.data._internal.table_block import TableBlockAccessor
  from ray.data.aggregate import AggregateFn, Count
  from ray.data.aggregate._aggregate import _AggregateOnKeyBase
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata, KeyType
label_desc: Use image node_id 1 to create a new SortAggregateTaskSpec object from inherited base classes: ExchangeTaskSpec with example: obj = SortAggregateTaskSpec(boundaries, key, aggs)