python/ray/data/grouped_dataset.py
def map(
        idx: int,
        block: Block,
        output_num_blocks: int,
        boundaries: List[KeyType],
        key: KeyFn,
        aggs: Tuple[AggregateFn],
    ) -> List[Union[BlockMetadata, Block]]:
        """Partition the block and combine rows with the same key."""
        stats = BlockExecStats.builder()
        if key is None:
            partitions = [block]
        else:
            partitions = BlockAccessor.for_block(block).sort_and_partition(
                boundaries,
                [(key, "ascending")] if isinstance(key, str) else key,
                descending=False,
            )
        parts = [BlockAccessor.for_block(p).combine(key, aggs) for p in partitions]
        meta = BlockAccessor.for_block(block).get_metadata(
            input_files=None, exec_stats=stats.build()
        )
        return [meta] + parts
Partition the block and combine rows with the same key.
Partition the block and combine rows with the same key.
Partition the block and combine rows with the same key .
def map(
        idx: int,
        block: Block,
        output_num_blocks: int,
        boundaries: List[KeyType],
        key: KeyFn,
        aggs: Tuple[AggregateFn],
    ) -> List[Union[BlockMetadata, Block]]:
        stats = BlockExecStats.builder()
        if key is None:
            partitions = [block]
        else:
            partitions = BlockAccessor.for_block(block).sort_and_partition(
                boundaries,
                [(key, "ascending")] if isinstance(key, str) else key,
                descending=False,
            )
        parts = [BlockAccessor.for_block(p).combine(key, aggs) for p in partitions]
        meta = BlockAccessor.for_block(block).get_metadata(
            input_files=None, exec_stats=stats.build()
        )
        return [meta] + parts
def map ( idx : int , block : Block , output_num_blocks : int , boundaries : List [ KeyType ] , key : KeyFn , aggs : Tuple [ AggregateFn ] , ) -> List [ Union [ BlockMetadata , Block ] ] : stats = BlockExecStats . builder ( ) if key is None : partitions = [ block ] else : partitions = BlockAccessor . for_block ( block ) . sort_and_partition ( boundaries , [ ( key , "ascending" ) ] if isinstance ( key , str ) else key , descending = False , ) parts = [ BlockAccessor . for_block ( p ) . combine ( key , aggs ) for p in partitions ] meta = BlockAccessor . for_block ( block ) . get_metadata ( input_files = None , exec_stats = stats . build ( ) ) return [ meta ] + parts
Partition the block and combine rows with the same key.
Partition the block and combine rows with the same key .
"""Partition the block and combine rows with the same key."""
idx
int
block
Block
output_num_blocks
int
boundaries
List[KeyType]
key
KeyFn
aggs
Tuple[AggregateFn]