python/ray/data/grouped_dataset.py
def aggregate(self, *aggs: AggregateFn) -> Dataset[U]:
        """Implements an accumulator-based aggregation.

        This is a blocking operation.

        Examples:
            >>> import ray
            >>> from ray.data.aggregate import AggregateFn
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> grouped_ds = ds.groupby(lambda x: x % 3) # doctest: +SKIP
            >>> grouped_ds.aggregate(AggregateFn( # doctest: +SKIP
            ...     init=lambda k: [], # doctest: +SKIP
            ...     accumulate=lambda a, r: a + [r], # doctest: +SKIP
            ...     merge=lambda a1, a2: a1 + a2, # doctest: +SKIP
            ...     finalize=lambda a: a # doctest: +SKIP
            ... )) # doctest: +SKIP

        Args:
            aggs: Aggregations to do.

        Returns:
            If the input dataset is simple dataset then the output is a simple
            dataset of ``(k, v_1, ..., v_n)`` tuples where ``k`` is the groupby
            key and ``v_i`` is the result of the ith given aggregation.
            If the input dataset is an Arrow dataset then the output is an
            Arrow 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, 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)
            # 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(),
                    [(self._key, "ascending")]
                    if isinstance(self._key, str)
                    else self._key,
                    num_reducers,
                )
            shuffle_op = GroupbyOp(
                map_args=[boundaries, self._key, aggs], reduce_args=[self._key, aggs]
            )
            return shuffle_op.execute(
                blocks,
                num_reducers,
                clear_input_blocks,
            )

        plan = self._dataset._plan.with_stage(AllToAllStage("aggregate", None, do_agg))
        return Dataset(
            plan,
            self._dataset._epoch,
            self._dataset._lazy,
        )
Implements an accumulator-based aggregation.

        This is a blocking operation.

        Examples:
            >>> import ray
            >>> from ray.data.aggregate import AggregateFn
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> grouped_ds = ds.groupby(lambda x: x % 3) # doctest: +SKIP
            >>> grouped_ds.aggregate(AggregateFn( # doctest: +SKIP
            ...     init=lambda k: [], # doctest: +SKIP
            ...     accumulate=lambda a, r: a + [r], # doctest: +SKIP
            ...     merge=lambda a1, a2: a1 + a2, # doctest: +SKIP
            ...     finalize=lambda a: a # doctest: +SKIP
            ... )) # doctest: +SKIP

        Args:
            aggs: Aggregations to do.

        Returns:
            If the input dataset is simple dataset then the output is a simple
            dataset of ``(k, v_1, ..., v_n)`` tuples where ``k`` is the groupby
            key and ``v_i`` is the result of the ith given aggregation.
            If the input dataset is an Arrow dataset then the output is an
            Arrow 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.
        
Implements an accumulator-based aggregation.
This is a blocking operation.
Implements an accumulator - based aggregation . This is a blocking operation .
def aggregate(self, *aggs: AggregateFn) -> Dataset[U]:
        def do_agg(blocks, clear_input_blocks: bool, *_):
            stage_info = {}
            if len(aggs) == 0:
                raise ValueError("Aggregate requires at least one aggregation")
            for agg in aggs:
                agg._validate(self._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(),
                    [(self._key, "ascending")]
                    if isinstance(self._key, str)
                    else self._key,
                    num_reducers,
                )
            shuffle_op = GroupbyOp(
                map_args=[boundaries, self._key, aggs], reduce_args=[self._key, aggs]
            )
            return shuffle_op.execute(
                blocks,
                num_reducers,
                clear_input_blocks,
            )
        plan = self._dataset._plan.with_stage(AllToAllStage("aggregate", None, do_agg))
        return Dataset(
            plan,
            self._dataset._epoch,
            self._dataset._lazy,
        )
def aggregate ( self , * aggs : AggregateFn ) -> Dataset [ U ] : def do_agg ( blocks , clear_input_blocks : bool , * _ ) : stage_info = { } if len ( aggs ) == 0 : raise ValueError ( "Aggregate requires at least one aggregation" ) for agg in aggs : agg . _validate ( self . _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 ( ) , [ ( self . _key , "ascending" ) ] if isinstance ( self . _key , str ) else self . _key , num_reducers , ) shuffle_op = GroupbyOp ( map_args = [ boundaries , self . _key , aggs ] , reduce_args = [ self . _key , aggs ] ) return shuffle_op . execute ( blocks , num_reducers , clear_input_blocks , ) plan = self . _dataset . _plan . with_stage ( AllToAllStage ( "aggregate" , None , do_agg ) ) return Dataset ( plan , self . _dataset . _epoch , self . _dataset . _lazy , )
Implements an accumulator-based aggregation.
Implements an accumulator - based aggregation .
"""Implements an accumulator-based aggregation.

        This is a blocking operation.

        Examples:
            >>> import ray
            >>> from ray.data.aggregate import AggregateFn
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> grouped_ds = ds.groupby(lambda x: x % 3) # doctest: +SKIP
            >>> grouped_ds.aggregate(AggregateFn( # doctest: +SKIP
            ...     init=lambda k: [], # doctest: +SKIP
            ...     accumulate=lambda a, r: a + [r], # doctest: +SKIP
            ...     merge=lambda a1, a2: a1 + a2, # doctest: +SKIP
            ...     finalize=lambda a: a # doctest: +SKIP
            ... )) # doctest: +SKIP

        Args:
            aggs: Aggregations to do.

        Returns:
            If the input dataset is simple dataset then the output is a simple
            dataset of ``(k, v_1, ..., v_n)`` tuples where ``k`` is the groupby
            key and ``v_i`` is the result of the ith given aggregation.
            If the input dataset is an Arrow dataset then the output is an
            Arrow 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.
        """ # TODO: implement clear_input_blocks # Handle empty dataset.
If the input dataset is simple dataset then the output is a simple
dataset of ``(k, v_1, ..., v_n)`` tuples where ``k`` is the groupby
key and ``v_i`` is the result of the ith given aggregation.
If the input dataset is an Arrow dataset then the output is an
Arrow 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.
If the input dataset is simple dataset then the output is a simple dataset of ` ` ( k v_1 ... v_n ) ` ` tuples where ` ` k ` ` is the groupby key and ` ` v_i ` ` is the result of the ith given aggregation . If the input dataset is an Arrow dataset then the output is an Arrow 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 .
self
aggs
AggregateFn
Aggregations to do.
Aggregations to do .
examples
>>> import ray
>>> from ray.data.aggregate import AggregateFn
>>> ds = ray.data.range(100) # doctest: +SKIP
>>> grouped_ds = ds.groupby(lambda x: x % 3) # doctest: +SKIP
>>> grouped_ds.aggregate(AggregateFn( # doctest: +SKIP
init=lambda k: [], # doctest: +SKIP
accumulate=lambda a, r: a + [r], # doctest: +SKIP
merge=lambda a1, a2: a1 + a2, # doctest: +SKIP
finalize=lambda a: a # doctest: +SKIP
)) # doctest: +SKIP
>>> import ray >>> from ray . data . aggregate import AggregateFn >>> ds = ray . data . range ( 100 ) # doctest : + SKIP >>> grouped_ds = ds . groupby ( lambda x : x % 3 ) # doctest : + SKIP >>> grouped_ds . aggregate ( AggregateFn ( # doctest : + SKIP init = lambda k : [] # doctest : + SKIP accumulate = lambda a r : a + [ r ] # doctest : + SKIP merge = lambda a1 a2 : a1 + a2 # doctest : + SKIP finalize = lambda a : a # doctest : + SKIP )) # doctest : + SKIP