docstr: The RayContext would initiate a ray cluster on top of the configuration of SparkContext.
  After creating RayContext, call the init method to set up the cluster.
  
  - For Spark local mode: The total available cores for Ray is equal to the number of
  Spark local cores.
  - For Spark cluster mode: The number of raylets to be created is equal to the number of
  Spark executors. The number of cores allocated for each raylet is equal to the number of
  cores for each Spark executor.
  You are allowed to specify num_ray_nodes and ray_node_cpu_cores for configurations
  to start raylets.
  
  :param sc: An instance of SparkContext.
  :param redis_port: The redis port for the ray head node. Default is None.
  The value would be randomly picked if not specified.
  :param redis_password: The password for redis. Default to be None if not specified.
  :param object_store_memory: The memory size for ray object_store in string.
  This can be specified in bytes(b), kilobytes(k), megabytes(m) or gigabytes(g).
  For example, "50b", "100k", "250m", "30g".
  :param verbose: True for more logs when starting ray. Default is False.
  :param env: The environment variable dict for running ray processes. Default is None.
  :param extra_params: The key value dict for extra options to launch ray.
  For example, extra_params={"temp-dir": "/tmp/ray/"}
  :param include_webui: True for including web ui when starting ray. Default is False.
  :param num_ray_nodes: The number of raylets to start across the cluster.
  For Spark local mode, you don't need to specify this value.
  For Spark cluster mode, it is default to be the number of Spark executors. If
  spark.executor.instances can't be detected in your SparkContext, you need to explicitly
  specify this. It is recommended that num_ray_nodes is not larger than the number of
  Spark executors to make sure there are enough resources in your cluster.
  :param ray_node_cpu_cores: The number of available cores for each raylet.
  For Spark local mode, it is default to be the number of Spark local cores.
  For Spark cluster mode, it is default to be the number of cores for each Spark executor. If
  spark.executor.cores or spark.cores.max can't be detected in your SparkContext, you need to
  explicitly specify this. It is recommended that ray_node_cpu_cores is not larger than the
  number of cores for each Spark executor to make sure there are enough resources in your
  cluster.
docstr_tok: ["The","RayContext","would","initiate","a","ray","cluster","on","top","of","the","configuration","of","SparkContext",".","After","creating","RayContext",",","call","the","init","method","to","set","up","the","cluster",".","-","For","Spark","local","mode",":","The","total","available","cores","for","Ray","is","equal","to","the","number","of","Spark","local","cores",".","-","For","Spark","cluster","mode",":","The","number","of","raylets","to","be","created","is","equal","to","the","number","of","Spark","executors",".","The","number","of","cores","allocated","for","each","raylet","is","equal","to","the","number","of","cores","for","each","Spark","executor",".","You","are","allowed","to","specify","num_ray_nodes","and","ray_node_cpu_cores","for","configurations","to","start","raylets",".",":","param","sc",":","An","instance","of","SparkContext",".",":","param","redis_port",":","The","redis","port","for","the","ray","head","node",".","Default","is","None",".","The","value","would","be","randomly","picked","if","not","specified",".",":","param","redis_password",":","The","password","for","redis",".","Default","to","be","None","if","not","specified",".",":","param","object_store_memory",":","The","memory","size","for","ray","object_store","in","string",".","This","can","be","specified","in","bytes","(","b",")",",","kilobytes","(","k",")",",","megabytes","(","m",")","or","gigabytes","(","g",")",".","For","example",",","``","50b","''",",","``","100k","''",",","``","250m","''",",","``","30g","''",".",":","param","verbose",":","True","for","more","logs","when","starting","ray",".","Default","is","False",".",":","param","env",":","The","environment","variable","dict","for","running","ray","processes",".","Default","is","None",".",":","param","extra_params",":","The","key","value","dict","for","extra","options","to","launch","ray",".","For","example",",","extra_params=","{","``","temp-dir","''",":","``","\/tmp\/ray\/","''","}",":","param","include_webui",":","True","for","including","web","ui","when","starting","ray",".","Default","is","False",".",":","param","num_ray_nodes",":","The","number","of","raylets","to","start","across","the","cluster",".","For","Spark","local","mode",",","you","do","n't","need","to","specify","this","value",".","For","Spark","cluster","mode",",","it","is","default","to","be","the","number","of","Spark","executors",".","If","spark.executor.instances","ca","n't","be","detected","in","your","SparkContext",",","you","need","to","explicitly","specify","this",".","It","is","recommended","that","num_ray_nodes","is","not","larger","than","the","number","of","Spark","executors","to","make","sure","there","are","enough","resources","in","your","cluster",".",":","param","ray_node_cpu_cores",":","The","number","of","available","cores","for","each","raylet",".","For","Spark","local","mode",",","it","is","default","to","be","the","number","of","Spark","local","cores",".","For","Spark","cluster","mode",",","it","is","default","to","be","the","number","of","cores","for","each","Spark","executor",".","If","spark.executor.cores","or","spark.cores.max","ca","n't","be","detected","in","your","SparkContext",",","you","need","to","explicitly","specify","this",".","It","is","recommended","that","ray_node_cpu_cores","is","not","larger","than","the","number","of","cores","for","each","Spark","executor","to","make","sure","there","are","enough","resources","in","your","cluster","."]
code: def __init__(
      self,
      sc,
      redis_port=None,
      redis_password=None,
      object_store_memory=None,
      verbose=False,
      env=None,
      extra_params=None,
      include_webui=True,
      num_ray_nodes=None,
      ray_node_cpu_cores=None,
  ):
      """
      The RayContext would initiate a ray cluster on top of the configuration of SparkContext.
      After creating RayContext, call the init method to set up the cluster.
  
      - For Spark local mode: The total available cores for Ray is equal to the number of
      Spark local cores.
      - For Spark cluster mode: The number of raylets to be created is equal to the number of
      Spark executors. The number of cores allocated for each raylet is equal to the number of
      cores for each Spark executor.
      You are allowed to specify num_ray_nodes and ray_node_cpu_cores for configurations
      to start raylets.
  
      :param sc: An instance of SparkContext.
      :param redis_port: The redis port for the ray head node. Default is None.
      The value would be randomly picked if not specified.
      :param redis_password: The password for redis. Default to be None if not specified.
      :param object_store_memory: The memory size for ray object_store in string.
      This can be specified in bytes(b), kilobytes(k), megabytes(m) or gigabytes(g).
      For example, "50b", "100k", "250m", "30g".
      :param verbose: True for more logs when starting ray. Default is False.
      :param env: The environment variable dict for running ray processes. Default is None.
      :param extra_params: The key value dict for extra options to launch ray.
      For example, extra_params={"temp-dir": "/tmp/ray/"}
      :param include_webui: True for including web ui when starting ray. Default is False.
      :param num_ray_nodes: The number of raylets to start across the cluster.
      For Spark local mode, you don't need to specify this value.
      For Spark cluster mode, it is default to be the number of Spark executors. If
      spark.executor.instances can't be detected in your SparkContext, you need to explicitly
      specify this. It is recommended that num_ray_nodes is not larger than the number of
      Spark executors to make sure there are enough resources in your cluster.
      :param ray_node_cpu_cores: The number of available cores for each raylet.
      For Spark local mode, it is default to be the number of Spark local cores.
      For Spark cluster mode, it is default to be the number of cores for each Spark executor. If
      spark.executor.cores or spark.cores.max can't be detected in your SparkContext, you need to
      explicitly specify this. It is recommended that ray_node_cpu_cores is not larger than the
      number of cores for each Spark executor to make sure there are enough resources in your
      cluster.
      """
      assert (
          sc is not None
      ), "sc cannot be None, please create a SparkContext first"
      self.sc = sc
      self.initialized = False
      self.is_local = is_local(sc)
      self.verbose = verbose
      self.redis_password = redis_password
      self.object_store_memory = resource_to_bytes(object_store_memory)
      self.ray_processesMonitor = None
      self.env = env
      self.extra_params = extra_params
      self.include_webui = include_webui
      self._address_info = None
      if self.is_local:
          self.num_ray_nodes = 1
          spark_cores = self._get_spark_local_cores()
          if ray_node_cpu_cores:
              ray_node_cpu_cores = int(ray_node_cpu_cores)
              if ray_node_cpu_cores > spark_cores:
                  warnings.warn(
                      "ray_node_cpu_cores is larger than available Spark cores, "
                      "make sure there are enough resources on your machine"
                  )
              self.ray_node_cpu_cores = ray_node_cpu_cores
          else:
              self.ray_node_cpu_cores = spark_cores
      # For Spark local mode, directly call ray.init() and ray.shutdown().
      # ray.shutdown() would clear up all the ray related processes.
      # Ray Manager is only needed for Spark cluster mode to monitor ray processes.
      else:
          if self.sc.getConf().contains("spark.executor.cores"):
              executor_cores = int(
                  self.sc.getConf().get("spark.executor.cores")
              )
          else:
              executor_cores = None
          if ray_node_cpu_cores:
              ray_node_cpu_cores = int(ray_node_cpu_cores)
              if executor_cores and ray_node_cpu_cores > executor_cores:
                  warnings.warn(
                      "ray_node_cpu_cores is larger than Spark executor cores, "
                      "make sure there are enough resources on your cluster"
                  )
              self.ray_node_cpu_cores = ray_node_cpu_cores
          elif executor_cores:
              self.ray_node_cpu_cores = executor_cores
          else:
              raise Exception(
                  "spark.executor.cores not detected in the SparkContext, "
                  "you need to manually specify num_ray_nodes and ray_node_cpu_cores "
                  "for RayContext to start ray services"
              )
          if self.sc.getConf().contains("spark.executor.instances"):
              num_executors = int(
                  self.sc.getConf().get("spark.executor.instances")
              )
          elif self.sc.getConf().contains("spark.cores.max"):
              import math
  
              num_executors = math.floor(
                  int(self.sc.getConf().get("spark.cores.max"))
                  / self.ray_node_cpu_cores
              )
          else:
              num_executors = None
          if num_ray_nodes:
              num_ray_nodes = int(num_ray_nodes)
              if num_executors and num_ray_nodes > num_executors:
                  warnings.warn(
                      "num_ray_nodes is larger than the number of Spark executors, "
                      "make sure there are enough resources on your cluster"
                  )
              self.num_ray_nodes = num_ray_nodes
          elif num_executors:
              self.num_ray_nodes = num_executors
          else:
              raise Exception(
                  "spark.executor.cores not detected in the SparkContext, "
                  "you need to manually specify num_ray_nodes and ray_node_cpu_cores "
                  "for RayContext to start ray services"
              )
  
          from zoo.util.utils import detect_python_location
  
          self.python_loc = os.environ.get(
              "PYSPARK_PYTHON", detect_python_location()
          )
          self.redis_port = (
              random.randint(10000, 65535)
              if not redis_port
              else int(redis_port)
          )
          self.ray_service = RayServiceFuncGenerator(
              python_loc=self.python_loc,
              redis_port=self.redis_port,
              ray_node_cpu_cores=self.ray_node_cpu_cores,
              redis_password=self.redis_password,
              object_store_memory=self.object_store_memory,
              verbose=self.verbose,
              env=self.env,
              include_webui=self.include_webui,
              extra_params=self.extra_params,
          )
      RayContext._active_ray_context = self
      self.total_cores = self.num_ray_nodes * self.ray_node_cpu_cores
  
imports: import os
  import re
  import subprocess
  import time
  import uuid
  import random
  import warnings
  import tempfile
  import filelock
  import multiprocessing
  from packaging import version
  from zoo.ray.process import session_execute, ProcessMonitor
  from zoo.ray.utils import is_local
  from zoo.ray.utils import resource_to_bytes
  from zoo.ray.utils import get_parent_pid
label_desc: Use image node_id 1 to create a new RayContext object from inherited base classes: object with example: obj = RayContext(sc, redis_port, redis_password, object_store_memory, verbose, env, extra_params, include_webui, num_ray_nodes, ray_node_cpu_cores)