python/ray/data/impl/simple_block.py
def combine(
        self, key: KeyFn, aggs: Tuple[AggregateFn]
    ) -> Block[Tuple[KeyType, AggType]]:
        """Combine rows with the same key into an accumulator.

        This assumes the block is already sorted by key in ascending order.

        Args:
            key: The key function that returns the key from the row
                or None for global aggregation.
            agg: The aggregations to do.

        Returns:
            A sorted block of (k, v_1, ..., v_n) tuples where k is the groupby
            key and v_i is the partially combined accumulator for the ith given
            aggregation.
            If key is None then the k element of tuple is omitted.
        """
        if key is not None and not callable(key):
            raise ValueError(
                "key must be a callable or None when aggregating on Simple blocks, but "
                f"got: {type(key)}."
            )

        def iter_groups() -> Iterator[Tuple[KeyType, Block]]:
            """Creates an iterator over zero-copy group views."""
            if key is None:
                # Global aggregation consists of a single "group", so we short-circuit.
                yield None, self.to_block()
                return

            start = end = 0
            iter = self.iter_rows()
            next_row = None
            # Use a bool to indicate if next_row is valid
            # instead of checking if next_row is None
            # since a row can have None value.
            has_next_row = False
            while True:
                try:
                    if not has_next_row:
                        next_row = next(iter)
                        has_next_row = True
                    next_key = key(next_row)
                    while key(next_row) == next_key:
                        end += 1
                        try:
                            next_row = next(iter)
                        except StopIteration:
                            has_next_row = False
                            next_row = None
                            break
                    yield next_key, self.slice(start, end, copy=False)
                    start = end
                except StopIteration:
                    break

        ret = []
        for group_key, group_view in iter_groups():
            # Aggregate.
            accumulators = [agg.init(group_key) for agg in aggs]
            for i in range(len(aggs)):
                accumulators[i] = aggs[i].accumulate_block(accumulators[i], group_view)

            # Build the row.
            if key is None:
                ret.append(tuple(accumulators))
            else:
                ret.append((group_key,) + tuple(accumulators))
        return ret
Combine rows with the same key into an accumulator.

        This assumes the block is already sorted by key in ascending order.

        Args:
            key: The key function that returns the key from the row
                or None for global aggregation.
            agg: The aggregations to do.

        Returns:
            A sorted block of (k, v_1, ..., v_n) tuples where k is the groupby
            key and v_i is the partially combined accumulator for the ith given
            aggregation.
            If key is None then the k element of tuple is omitted.
        
Combine rows with the same key into an accumulator.
This assumes the block is already sorted by key in ascending order.
Combine rows with the same key into an accumulator . This assumes the block is already sorted by key in ascending order .
def combine(
        self, key: KeyFn, aggs: Tuple[AggregateFn]
    ) -> Block[Tuple[KeyType, AggType]]:
        if key is not None and not callable(key):
            raise ValueError(
                "key must be a callable or None when aggregating on Simple blocks, but "
                f"got: {type(key)}."
            )
        def iter_groups() -> Iterator[Tuple[KeyType, Block]]:
            if key is None:
                yield None, self.to_block()
                return
            start = end = 0
            iter = self.iter_rows()
            next_row = None
            has_next_row = False
            while True:
                try:
                    if not has_next_row:
                        next_row = next(iter)
                        has_next_row = True
                    next_key = key(next_row)
                    while key(next_row) == next_key:
                        end += 1
                        try:
                            next_row = next(iter)
                        except StopIteration:
                            has_next_row = False
                            next_row = None
                            break
                    yield next_key, self.slice(start, end, copy=False)
                    start = end
                except StopIteration:
                    break
        ret = []
        for group_key, group_view in iter_groups():
            accumulators = [agg.init(group_key) for agg in aggs]
            for i in range(len(aggs)):
                accumulators[i] = aggs[i].accumulate_block(accumulators[i], group_view)
            if key is None:
                ret.append(tuple(accumulators))
            else:
                ret.append((group_key,) + tuple(accumulators))
        return ret
def combine ( self , key : KeyFn , aggs : Tuple [ AggregateFn ] ) -> Block [ Tuple [ KeyType , AggType ] ] : if key is not None and not callable ( key ) : raise ValueError ( "key must be a callable or None when aggregating on Simple blocks, but " f"got: {type(key)}." ) def iter_groups ( ) -> Iterator [ Tuple [ KeyType , Block ] ] : """Creates an iterator over zero-copy group views.""" if key is None : yield None , self . to_block ( ) return start = end = 0 iter = self . iter_rows ( ) next_row = None has_next_row = False while True : try : if not has_next_row : next_row = next ( iter ) has_next_row = True next_key = key ( next_row ) while key ( next_row ) == next_key : end += 1 try : next_row = next ( iter ) except StopIteration : has_next_row = False next_row = None break yield next_key , self . slice ( start , end , copy = False ) start = end except StopIteration : break ret = [ ] for group_key , group_view in iter_groups ( ) : accumulators = [ agg . init ( group_key ) for agg in aggs ] for i in range ( len ( aggs ) ) : accumulators [ i ] = aggs [ i ] . accumulate_block ( accumulators [ i ] , group_view ) if key is None : ret . append ( tuple ( accumulators ) ) else : ret . append ( ( group_key , ) + tuple ( accumulators ) ) return ret
Combine rows with the same key into an accumulator.
Combine rows with the same key into an accumulator .
"""Combine rows with the same key into an accumulator.

        This assumes the block is already sorted by key in ascending order.

        Args:
            key: The key function that returns the key from the row
                or None for global aggregation.
            agg: The aggregations to do.

        Returns:
            A sorted block of (k, v_1, ..., v_n) tuples where k is the groupby
            key and v_i is the partially combined accumulator for the ith given
            aggregation.
            If key is None then the k element of tuple is omitted.
        """ """Creates an iterator over zero-copy group views.""" # Global aggregation consists of a single "group", so we short-circuit. # Use a bool to indicate if next_row is valid # instead of checking if next_row is None # since a row can have None value. # Aggregate. # Build the row.
A sorted block of (k, v_1, ..., v_n) tuples where k is the groupby
key and v_i is the partially combined accumulator for the ith given
aggregation.
If key is None then the k element of tuple is omitted.
A sorted block of ( k v_1 ... v_n ) tuples where k is the groupby key and v_i is the partially combined accumulator for the ith given aggregation . If key is None then the k element of tuple is omitted .
self
key
KeyFn
The key function that returns the key from the row
or None for global aggregation.
The key function that returns the key from the row or None for global aggregation .
aggs
Tuple[AggregateFn]
agg
The aggregations to do.
The aggregations to do .