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 __init__(self):
      # execution_id -> (List[Dict], expiration timestamp)
      self._resource_requests = {}
      # TTL for requests.
      self._timeout = RESOURCE_REQUEST_TIMEOUT
  
      self._self_handle = ray.get_runtime_context().current_actor
  
      # Start a thread to purge expired requests periodically.
      def purge_thread_run():
          while True:
              time.sleep(PURGE_INTERVAL)
              # Call purge_expired_requests() as an actor task,
              # so we don't need to handle multi-threading.
              ray.get(self._self_handle.purge_expired_requests.remote())
  
      self._purge_thread = threading.Thread(
          target=purge_thread_run, daemon=True
      )
      self._purge_thread.start()
  
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 1 to create a new AutoscalingRequester object  with example: obj = AutoscalingRequester()