docstr: The RayOnSparkContext would initiate a ray cluster on top of the configuration of
  SparkContext.
  After creating RayOnSparkContext, 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={"dashboard-port": "11281", "temp-dir": "/tmp/ray/"}.
  :param include_webui: Default is True for including web ui when starting ray.
  :param num_ray_nodes: The number of ray processes 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 ray process.
  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.
  :param system_config: The key value dict for overriding RayConfig defaults. Mainly for
  testing purposes. An example for system_config could be:
  {"object_spilling_config":"{"type":"filesystem",
                             "params":{"directory_path":"/tmp/spill"}}"}
docstr_tok: ["The","RayOnSparkContext","would","initiate","a","ray","cluster","on","top","of","the","configuration","of","SparkContext",".","After","creating","RayOnSparkContext",",","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=","{","``","dashboard-port","''",":","``","11281","''",",","``","temp-dir","''",":","``","\/tmp\/ray\/","''","}",".",":","param","include_webui",":","Default","is","True","for","including","web","ui","when","starting","ray",".",":","param","num_ray_nodes",":","The","number","of","ray","processes","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","ray","process",".","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",".",":","param","system_config",":","The","key","value","dict","for","overriding","RayConfig","defaults",".","Mainly","for","testing","purposes",".","An","example","for","system_config","could","be",":","{","``","object_spilling_config","''",":","''","{","``","type","''",":","''","filesystem","''",",","``","params","''",":","{","``","directory_path","''",":","''","\/tmp\/spill","''","}","}","''","}"]
code: def __init__(
      self,
      sc: "SparkContext",
      redis_port: Optional[int] = None,
      redis_password: Optional[str] = None,
      object_store_memory: Optional[str] = None,
      verbose: bool = False,
      env: Optional[Dict[str, str]] = None,
      extra_params: Optional[Dict[str, Any]] = None,
      include_webui: bool = True,
      num_ray_nodes: Optional[int] = None,
      ray_node_cpu_cores: Optional[int] = None,
      system_config: Optional[Dict[str, str]] = None,
  ):
      """
      The RayOnSparkContext would initiate a ray cluster on top of the configuration of
      SparkContext.
      After creating RayOnSparkContext, 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={"dashboard-port": "11281", "temp-dir": "/tmp/ray/"}.
      :param include_webui: Default is True for including web ui when starting ray.
      :param num_ray_nodes: The number of ray processes 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 ray process.
      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.
      :param system_config: The key value dict for overriding RayConfig defaults. Mainly for
      testing purposes. An example for system_config could be:
      {"object_spilling_config":"{\"type\":\"filesystem\",
                                 \"params\":{\"directory_path\":\"/tmp/spill\"}}"}
      """
      invalidInputError(
          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.system_config = system_config
      if extra_params:
          invalidInputError(
              isinstance(extra_params, dict),
              "extra_params should be a dict for extra options to launch ray",
          )
          if self.system_config:
              self.extra_params.pop("system_config", None)  # type: ignore
              self.extra_params.pop("_system_config", None)  # type: ignore
          elif "system_config" in self.extra_params:
              self.system_config = self.extra_params.pop("system_config")  # type: ignore
          elif "_system_config" in self.extra_params:
              self.system_config = self.extra_params.pop("_system_config")  # type: ignore
      self.include_webui = include_webui
      self._address_info = None
      self.redis_port = (
          random.randint(20000, 65535)
          if not redis_port
          else int(redis_port)
      )
      self.ray_node_cpu_cores = ray_node_cpu_cores
      self.num_ray_nodes = num_ray_nodes
      RayOnSparkContext._active_ray_context = self
  
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 bigdl.orca.ray.process import session_execute, ProcessMonitor
  from bigdl.orca.ray.utils import is_local
  from bigdl.orca.ray.utils import resource_to_bytes
  from bigdl.orca.ray.utils import get_parent_pid
  from bigdl.dllib.utils.log4Error import invalidInputError
  from typing import TYPE_CHECKING
  from typing import Any, Dict, Optional, List
label_desc: Use image node_id 1 to create a new RayOnSparkContext object from inherited base classes: object with example: obj = RayOnSparkContext(sc, redis_port, redis_password, object_store_memory, verbose, env, extra_params, include_webui, num_ray_nodes, ray_node_cpu_cores, system_config)