python/ray/data/dataset.py
def repartition(self, num_blocks: int, *, shuffle: bool = False) -> "Dataset[T]":
        """Repartition the dataset into exactly this number of blocks.

        This is a blocking operation. After repartitioning, all blocks in the
        returned dataset will have approximately the same number of rows.

        Examples:
            >>> import ray
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> # Set the number of output partitions to write to disk.
            >>> ds.repartition(10).write_parquet(...) # doctest: +SKIP

        Time complexity: O(dataset size / parallelism)

        Args:
            num_blocks: The number of blocks.
            shuffle: Whether to perform a distributed shuffle during the
                repartition. When shuffle is enabled, each output block
                contains a subset of data rows from each input block, which
                requires all-to-all data movement. When shuffle is disabled,
                output blocks are created from adjacent input blocks,
                minimizing data movement.

        Returns:
            The repartitioned dataset.
        """

        if shuffle:

            def do_shuffle(
                block_list, clear_input_blocks: bool, block_udf, remote_args
            ):
                if clear_input_blocks:
                    blocks = block_list.copy()
                    block_list.clear()
                else:
                    blocks = block_list
                shuffle_op = ShufflePartitionOp(block_udf, random_shuffle=False)
                return shuffle_op.execute(
                    blocks,
                    num_blocks,
                    clear_input_blocks,
                    map_ray_remote_args=remote_args,
                    reduce_ray_remote_args=remote_args,
                )

            plan = self._plan.with_stage(
                AllToAllStage(
                    "repartition", num_blocks, do_shuffle, supports_block_udf=True
                )
            )

        else:

            def do_fast_repartition(block_list, clear_input_blocks: bool, *_):
                if clear_input_blocks:
                    blocks = block_list.copy()
                    block_list.clear()
                else:
                    blocks = block_list
                return fast_repartition(blocks, num_blocks)

            plan = self._plan.with_stage(
                AllToAllStage("repartition", num_blocks, do_fast_repartition)
            )

        return Dataset(plan, self._epoch, self._lazy)
Repartition the dataset into exactly this number of blocks.

        This is a blocking operation. After repartitioning, all blocks in the
        returned dataset will have approximately the same number of rows.

        Examples:
            >>> import ray
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> # Set the number of output partitions to write to disk.
            >>> ds.repartition(10).write_parquet(...) # doctest: +SKIP

        Time complexity: O(dataset size / parallelism)

        Args:
            num_blocks: The number of blocks.
            shuffle: Whether to perform a distributed shuffle during the
                repartition. When shuffle is enabled, each output block
                contains a subset of data rows from each input block, which
                requires all-to-all data movement. When shuffle is disabled,
                output blocks are created from adjacent input blocks,
                minimizing data movement.

        Returns:
            The repartitioned dataset.
        
Repartition the dataset into exactly this number of blocks.
This is a blocking operation. After repartitioning, all blocks in the
returned dataset will have approximately the same number of rows.
Repartition the dataset into exactly this number of blocks . This is a blocking operation . After repartitioning all blocks in the returned dataset will have approximately the same number of rows .
def repartition(self, num_blocks: int, *, shuffle: bool = False) -> "Dataset[T]":
        if shuffle:
            def do_shuffle(
                block_list, clear_input_blocks: bool, block_udf, remote_args
            ):
                if clear_input_blocks:
                    blocks = block_list.copy()
                    block_list.clear()
                else:
                    blocks = block_list
                shuffle_op = ShufflePartitionOp(block_udf, random_shuffle=False)
                return shuffle_op.execute(
                    blocks,
                    num_blocks,
                    clear_input_blocks,
                    map_ray_remote_args=remote_args,
                    reduce_ray_remote_args=remote_args,
                )
            plan = self._plan.with_stage(
                AllToAllStage(
                    "repartition", num_blocks, do_shuffle, supports_block_udf=True
                )
            )
        else:
            def do_fast_repartition(block_list, clear_input_blocks: bool, *_):
                if clear_input_blocks:
                    blocks = block_list.copy()
                    block_list.clear()
                else:
                    blocks = block_list
                return fast_repartition(blocks, num_blocks)
            plan = self._plan.with_stage(
                AllToAllStage("repartition", num_blocks, do_fast_repartition)
            )
        return Dataset(plan, self._epoch, self._lazy)
def repartition ( self , num_blocks : int , * , shuffle : bool = False ) -> "Dataset[T]" : if shuffle : def do_shuffle ( block_list , clear_input_blocks : bool , block_udf , remote_args ) : if clear_input_blocks : blocks = block_list . copy ( ) block_list . clear ( ) else : blocks = block_list shuffle_op = ShufflePartitionOp ( block_udf , random_shuffle = False ) return shuffle_op . execute ( blocks , num_blocks , clear_input_blocks , map_ray_remote_args = remote_args , reduce_ray_remote_args = remote_args , ) plan = self . _plan . with_stage ( AllToAllStage ( "repartition" , num_blocks , do_shuffle , supports_block_udf = True ) ) else : def do_fast_repartition ( block_list , clear_input_blocks : bool , * _ ) : if clear_input_blocks : blocks = block_list . copy ( ) block_list . clear ( ) else : blocks = block_list return fast_repartition ( blocks , num_blocks ) plan = self . _plan . with_stage ( AllToAllStage ( "repartition" , num_blocks , do_fast_repartition ) ) return Dataset ( plan , self . _epoch , self . _lazy )
Repartition the dataset into exactly this number of blocks.
Repartition the dataset into exactly this number of blocks .
"""Repartition the dataset into exactly this number of blocks.

        This is a blocking operation. After repartitioning, all blocks in the
        returned dataset will have approximately the same number of rows.

        Examples:
            >>> import ray
            >>> ds = ray.data.range(100) # doctest: +SKIP
            >>> # Set the number of output partitions to write to disk.
            >>> ds.repartition(10).write_parquet(...) # doctest: +SKIP

        Time complexity: O(dataset size / parallelism)

        Args:
            num_blocks: The number of blocks.
            shuffle: Whether to perform a distributed shuffle during the
                repartition. When shuffle is enabled, each output block
                contains a subset of data rows from each input block, which
                requires all-to-all data movement. When shuffle is disabled,
                output blocks are created from adjacent input blocks,
                minimizing data movement.

        Returns:
            The repartitioned dataset.
        """
The repartitioned dataset.
The repartitioned dataset .
self
num_blocks
int
The number of blocks.
The number of blocks .
shuffle
bool
Whether to perform a distributed shuffle during the
repartition. When shuffle is enabled, each output block
contains a subset of data rows from each input block, which
requires all-to-all data movement. When shuffle is disabled,
output blocks are created from adjacent input blocks,
minimizing data movement.
Whether to perform a distributed shuffle during the repartition . When shuffle is enabled each output block contains a subset of data rows from each input block which requires all - to - all data movement . When shuffle is disabled output blocks are created from adjacent input blocks minimizing data movement .
examples
>>> import ray
>>> ds = ray.data.range(100) # doctest: +SKIP
>>> # Set the number of output partitions to write to disk.
>>> ds.repartition(10).write_parquet(...) # doctest: +SKIP
>>> import ray >>> ds = ray . data . range ( 100 ) # doctest : + SKIP >>> # Set the number of output partitions to write to disk . >>> ds . repartition ( 10 ) . write_parquet ( ... ) # doctest : + SKIP