python/ray/data/impl/sort.py
def sample_boundaries(
    blocks: List[ObjectRef[Block]], key: SortKeyT, num_reducers: int
) -> List[T]:
    """
    Return (num_reducers - 1) items in ascending order from the blocks that
    partition the domain into ranges with approximately equally many elements.
    """
    # TODO(Clark): Support multiple boundary sampling keys.
    if isinstance(key, list) and len(key) > 1:
        raise ValueError("Multiple boundary sampling keys not supported.")

    n_samples = int(num_reducers * 10 / len(blocks))

    sample_block = cached_remote_fn(_sample_block)

    sample_results = [sample_block.remote(block, n_samples, key) for block in blocks]
    sample_bar = ProgressBar("Sort Sample", len(sample_results))
    sample_bar.block_until_complete(sample_results)
    sample_bar.close()

    samples = ray.get(sample_results)
    samples = [s for s in samples if len(s) > 0]
    # The dataset is empty
    if len(samples) == 0:
        return [None] * (num_reducers - 1)
    builder = DelegatingBlockBuilder()
    for sample in samples:
        builder.add_block(sample)
    samples = builder.build()
    column = key[0][0] if isinstance(key, list) else None
    sample_items = BlockAccessor.for_block(samples).to_numpy(column)
    sample_items = np.sort(sample_items)
    ret = [
        np.quantile(sample_items, q, interpolation="nearest")
        for q in np.linspace(0, 1, num_reducers)
    ]
    return ret[1:]

    Return (num_reducers - 1) items in ascending order from the blocks that
    partition the domain into ranges with approximately equally many elements.
    
Return (num_reducers - 1) items in ascending order from the blocks that
partition the domain into ranges with approximately equally many elements.
Return ( num_reducers - 1 ) items in ascending order from the blocks that partition the domain into ranges with approximately equally many elements .
def sample_boundaries(
    blocks: List[ObjectRef[Block]], key: SortKeyT, num_reducers: int
) -> List[T]:
    if isinstance(key, list) and len(key) > 1:
        raise ValueError("Multiple boundary sampling keys not supported.")
    n_samples = int(num_reducers * 10 / len(blocks))
    sample_block = cached_remote_fn(_sample_block)
    sample_results = [sample_block.remote(block, n_samples, key) for block in blocks]
    sample_bar = ProgressBar("Sort Sample", len(sample_results))
    sample_bar.block_until_complete(sample_results)
    sample_bar.close()
    samples = ray.get(sample_results)
    samples = [s for s in samples if len(s) > 0]
    if len(samples) == 0:
        return [None] * (num_reducers - 1)
    builder = DelegatingBlockBuilder()
    for sample in samples:
        builder.add_block(sample)
    samples = builder.build()
    column = key[0][0] if isinstance(key, list) else None
    sample_items = BlockAccessor.for_block(samples).to_numpy(column)
    sample_items = np.sort(sample_items)
    ret = [
        np.quantile(sample_items, q, interpolation="nearest")
        for q in np.linspace(0, 1, num_reducers)
    ]
    return ret[1:]
def sample_boundaries ( blocks : List [ ObjectRef [ Block ] ] , key : SortKeyT , num_reducers : int ) -> List [ T ] : if isinstance ( key , list ) and len ( key ) > 1 : raise ValueError ( "Multiple boundary sampling keys not supported." ) n_samples = int ( num_reducers * 10 / len ( blocks ) ) sample_block = cached_remote_fn ( _sample_block ) sample_results = [ sample_block . remote ( block , n_samples , key ) for block in blocks ] sample_bar = ProgressBar ( "Sort Sample" , len ( sample_results ) ) sample_bar . block_until_complete ( sample_results ) sample_bar . close ( ) samples = ray . get ( sample_results ) samples = [ s for s in samples if len ( s ) > 0 ] if len ( samples ) == 0 : return [ None ] * ( num_reducers - 1 ) builder = DelegatingBlockBuilder ( ) for sample in samples : builder . add_block ( sample ) samples = builder . build ( ) column = key [ 0 ] [ 0 ] if isinstance ( key , list ) else None sample_items = BlockAccessor . for_block ( samples ) . to_numpy ( column ) sample_items = np . sort ( sample_items ) ret = [ np . quantile ( sample_items , q , interpolation = "nearest" ) for q in np . linspace ( 0 , 1 , num_reducers ) ] return ret [ 1 : ]
Return (num_reducers - 1) items in ascending order from the blocks that
partition the domain into ranges with approximately equally many elements.
Return ( num_reducers - 1 ) items in ascending order from the blocks that partition the domain into ranges with approximately equally many elements .
"""
    Return (num_reducers - 1) items in ascending order from the blocks that
    partition the domain into ranges with approximately equally many elements.
    """ # TODO(Clark): Support multiple boundary sampling keys. # The dataset is empty
blocks
List[ObjectRef[Block]]
key
SortKeyT
num_reducers
int