class_docstr: The implementation for shuffle tasks.
  
  This is used by random_shuffle() and repartition().
class_docstr_tok: ["The","implementation","for","shuffle","tasks",".","This","is","used","by","random_shuffle","(",")","and","repartition","(",")","."]
code: def reduce(
      random_shuffle: bool,
      random_seed: Optional[int],
      *mapper_outputs: List[Block],
      partial_reduce: bool = False,
  ) -> Tuple[Block, BlockMetadata]:
      # TODO: Support fusion with other downstream operators.
      stats = BlockExecStats.builder()
      builder = DelegatingBlockBuilder()
      for block in mapper_outputs:
          builder.add_block(block)
      new_block = builder.build()
      accessor = BlockAccessor.for_block(new_block)
      if random_shuffle:
          new_block = accessor.random_shuffle(
              random_seed if random_seed is not None else None
          )
          accessor = BlockAccessor.for_block(new_block)
      new_metadata = BlockMetadata(
          num_rows=accessor.num_rows(),
          size_bytes=accessor.size_bytes(),
          schema=accessor.schema(),
          input_files=None,
          exec_stats=stats.build(),
      )
      return new_block, new_metadata
  
file_path: ray/python/ray/data/_internal/planner/exchange/shuffle_task_spec.py
imports: import math
  from typing import Callable, Iterable, List, Optional, Tuple, Union
  import numpy
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data._internal.planner.exchange.interfaces import ExchangeTaskSpec
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata
  from ray.data.context import MAX_SAFE_BLOCK_SIZE_FACTOR
label_desc: Use image node_id 3 for calling the ShuffleTaskSpec obj's underlying member method code with example usage: obj.reduce(random_shuffle, random_seed) and returns: new_block,  new_metadata