class_docstr: A node provider that implements provider for nodes of Ray on spark.
class_docstr_tok: ["A","node","provider","that","implements","provider","for","nodes","of","Ray","on","spark","."]
code: def __init__(self, provider_config, cluster_name):
      NodeProvider.__init__(self, provider_config, cluster_name)
      self.lock = RLock()
  
      self._nodes = {
          str(HEAD_NODE_ID): {
              "tags": {
                  TAG_RAY_NODE_KIND: NODE_KIND_HEAD,
                  TAG_RAY_USER_NODE_TYPE: HEAD_NODE_TYPE,
                  TAG_RAY_NODE_NAME: HEAD_NODE_ID,
                  TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE,
              }
          },
      }
      self._next_node_id = 0
  
      self.ray_head_ip = self.provider_config["ray_head_ip"]
      # The port of spark job server. We send http request to spark job server
      # to launch spark jobs, ray worker nodes are launched by spark task in
      # spark jobs.
      spark_job_server_port = self.provider_config[
          "spark_job_server_port"
      ]
      self.spark_job_server_url = (
          f"http://{self.ray_head_ip}:{spark_job_server_port}"
      )
      self.ray_head_port = self.provider_config["ray_head_port"]
      # The unique id for the Ray on spark cluster.
      self.cluster_id = self.provider_config["cluster_unique_id"]
  
imports: import json
  import logging
  import sys
  from threading import RLock
  from typing import Any, Dict, Optional
  import requests
  from ray.autoscaler.node_launch_exception import NodeLaunchException
  from ray.autoscaler.node_provider import NodeProvider
  from ray.autoscaler.tags import NODE_KIND_HEAD, NODE_KIND_WORKER, STATUS_SETTING_UP, STATUS_UP_TO_DATE, TAG_RAY_NODE_KIND, TAG_RAY_NODE_NAME, TAG_RAY_NODE_STATUS, TAG_RAY_USER_NODE_TYPE
label_desc: Use image node_id 1 to create a new SparkNodeProvider object from inherited base classes: NodeProvider with example: obj = SparkNodeProvider(provider_config, cluster_name)