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 map(
      idx: int,
      block: Block,
      output_num_blocks: int,
      target_shuffle_max_block_size: int,
      upstream_map_fn: Optional[
          Callable[[Iterable[Block]], Iterable[Block]]
      ],
      random_shuffle: bool,
      random_seed: Optional[int],
  ) -> List[Union[BlockMetadata, Block]]:
      stats = BlockExecStats.builder()
      if upstream_map_fn:
          # TODO: Support dynamic block splitting in
          # all-to-all ops, to avoid having to re-fuse
          # upstream blocks together.
          upstream_map_iter = upstream_map_fn([block])
          mapped_block = next(upstream_map_iter)
          builder = BlockAccessor.for_block(mapped_block).builder()
          builder.add_block(mapped_block)
          for mapped_block in upstream_map_iter:
              builder.add_block(mapped_block)
          # Drop the upstream inputs to reduce memory usage.
          del mapped_block
          block = builder.build()
      block = BlockAccessor.for_block(block)
      if (
          block.size_bytes()
          > MAX_SAFE_BLOCK_SIZE_FACTOR * target_shuffle_max_block_size
      ):
          logger.get_logger().warn(
              "Input block to map task has size "
              f"{block.size_bytes() // (1024 * 1024)}MiB, which exceeds "
              "DataContext.get_current().target_shuffle_max_block_size="
              f"{target_shuffle_max_block_size // (1024 * 1024)}MiB. "
              "This can lead to out-of-memory errors and can happen "
              "when map tasks are fused to the shuffle operation. "
              "To prevent fusion, call Dataset.materialize() on the "
              "dataset before shuffling."
          )
  
      # Randomize the distribution of records to blocks.
      if random_shuffle:
          seed_i = (
              random_seed + idx if random_seed is not None else None
          )
          block = block.random_shuffle(seed_i)
          block = BlockAccessor.for_block(block)
  
      # Build a list of slices to return. It's okay to put the results in a
      # list instead of yielding them as a generator because slicing the
      # ArrowBlock is zero-copy.
      slice_sz = max(1, math.ceil(block.num_rows() / output_num_blocks))
      slices = []
      for i in range(output_num_blocks):
          slices.append(block.slice(i * slice_sz, (i + 1) * slice_sz))
  
      # Randomize the distribution order of the blocks (this prevents empty
      # outputs when input blocks are very small).
      if random_shuffle:
          random = np.random.RandomState(seed_i)
          random.shuffle(slices)
  
      num_rows = sum(
          BlockAccessor.for_block(s).num_rows() for s in slices
      )
      assert num_rows == block.num_rows(), (num_rows, block.num_rows())
      metadata = block.get_metadata(
          input_files=None, exec_stats=stats.build()
      )
      return slices + [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 2 for calling the ShuffleTaskSpec obj's underlying member method code with example usage: obj.map(idx, block, output_num_blocks, target_shuffle_max_block_size, upstream_map_fn, random_shuffle, random_seed) and returns: unknown