class_docstr: Actor to make resource requests to autoscaler for the datasets.
  
  The resource requests are set to timeout after RESOURCE_REQUEST_TIMEOUT seconds.
  For those live requests, we keep track of the last request made for each execution,
  which overrides all previous requests it made; then sum the requested amounts
  across all executions as the final request to the autoscaler.
code: def _aggregate_requests(self) -> List[Dict]:
      req = []
      for _, (r, _) in self._resource_requests.items():
          req.extend(r)
  
      def get_cpus(req):
          num_cpus = 0
          for r in req:
              if "CPU" in r:
                  num_cpus += r["CPU"]
          return num_cpus
  
      # Round up CPUs to exceed total cluster CPUs so it can actually upscale.
      # This is to handle the issue where the autoscaling is driven by memory
      # pressure (rather than CPUs) from streaming executor. In such case, simply
      # asking for incremental CPUs (e.g. 1 CPU for each ready operator) may not
      # actually be able to trigger autoscaling if existing CPUs in cluster can
      # already satisfy the incremental CPUs request.
      num_cpus = get_cpus(req)
      if num_cpus > 0:
          total = ray.cluster_resources()
          if "CPU" in total and num_cpus <= total["CPU"]:
              delta = (
                  math.ceil(
                      ARTIFICIAL_CPU_SCALING_FACTOR * total["CPU"]
                  )
                  - num_cpus
              )
              req.extend([{"CPU": 1}] * delta)
  
      return req
  
file_path: ray/python/ray/data/_internal/execution/autoscaling_requester.py
imports: import math
  import threading
  import time
  from typing import Dict, List
  import ray
  from ray.data.context import DataContext
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 5 for calling the AutoscalingRequester obj's underlying member method code with example usage: obj._aggregate_requests() and returns: req, num_cpus