class_docstr: A class that provides distributed, random access to a Dataset.
  
  See: ``Dataset.to_random_access_dataset()``.
class_docstr_tok: ["A","class","that","provides","distributed",",","random","access","to","a","Dataset",".","See",":","``","Dataset.to_random_access_dataset","(",")","``","."]
docstr: Construct a RandomAccessDataset (internal API).
  
  The constructor is a private API. Use ``ds.to_random_access_dataset()``
  to construct a RandomAccessDataset.
docstr_tok: ["Construct","a","RandomAccessDataset","(","internal","API",")",".","The","constructor","is","a","private","API",".","Use","``","ds.to_random_access_dataset","(",")","``","to","construct","a","RandomAccessDataset","."]
code: def __init__(
      self,
      ds: "Dataset",
      key: str,
      num_workers: int,
  ):
      """Construct a RandomAccessDataset (internal API).
  
      The constructor is a private API. Use ``ds.to_random_access_dataset()``
      to construct a RandomAccessDataset.
      """
      schema = ds.schema(fetch_if_missing=True)
      if schema is None or isinstance(schema, type):
          raise ValueError(
              "RandomAccessDataset only supports Arrow-format blocks."
          )
  
      start = time.perf_counter()
      logger.info("[setup] Indexing dataset by sort key.")
      sorted_ds = ds.sort(key)
      get_bounds = cached_remote_fn(_get_bounds)
      blocks = sorted_ds.get_internal_block_refs()
  
      logger.info("[setup] Computing block range bounds.")
      bounds = ray.get([get_bounds.remote(b, key) for b in blocks])
      self._non_empty_blocks = []
      self._lower_bound = None
      self._upper_bounds = []
      for i, b in enumerate(bounds):
          if b:
              self._non_empty_blocks.append(blocks[i])
              if self._lower_bound is None:
                  self._lower_bound = b[0]
              self._upper_bounds.append(b[1])
  
      logger.info(
          "[setup] Creating {} random access workers.".format(
              num_workers
          )
      )
      ctx = DataContext.get_current()
      scheduling_strategy = ctx.scheduling_strategy
      self._workers = [
          _RandomAccessWorker.options(
              scheduling_strategy=scheduling_strategy
          ).remote(key)
          for _ in range(num_workers)
      ]
      (
          self._block_to_workers_map,
          self._worker_to_blocks_map,
      ) = self._compute_block_to_worker_assignments()
  
      logger.info(
          "[setup] Worker to blocks assignment: {}".format(
              self._worker_to_blocks_map
          )
      )
      ray.get(
          [
              w.assign_blocks.remote(
                  {
                      i: self._non_empty_blocks[i]
                      for i in self._worker_to_blocks_map[w]
                  }
              )
              for w in self._workers
          ]
      )
  
      logger.info("[setup] Finished assigning blocks to workers.")
      self._build_time = time.perf_counter() - start
  
imports: import bisect
  import logging
  import random
  import time
  from collections import defaultdict
  from typing import TYPE_CHECKING, Any, List, Optional
  import numpy
  import ray
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 1 to create a new RandomAccessDataset object  with example: obj = RandomAccessDataset(ds, key, num_workers)