docstr: Initiate the ray cluster.
  
  :param driver_cores: The number of cores for the raylet on driver for Spark cluster mode.
  Default is 0 and in this case the local driver wouldn't have any ray workload.
  
  :return The dictionary of address information about the ray cluster.
  Information contains node_ip_address, redis_address, object_store_address,
  raylet_socket_name, webui_url and session_dir.
docstr_tok: ["Initiate","the","ray","cluster",".",":","param","driver_cores",":","The","number","of","cores","for","the","raylet","on","driver","for","Spark","cluster","mode",".","Default","is","0","and","in","this","case","the","local","driver","would","n't","have","any","ray","workload",".",":","return","The","dictionary","of","address","information","about","the","ray","cluster",".","Information","contains","node_ip_address",",","redis_address",",","object_store_address",",","raylet_socket_name",",","webui_url","and","session_dir","."]
code: def init(self, driver_cores=0):
      """
      Initiate the ray cluster.
  
      :param driver_cores: The number of cores for the raylet on driver for Spark cluster mode.
      Default is 0 and in this case the local driver wouldn't have any ray workload.
  
      :return The dictionary of address information about the ray cluster.
      Information contains node_ip_address, redis_address, object_store_address,
      raylet_socket_name, webui_url and session_dir.
      """
      if self.initialized:
          print("The Ray cluster has been launched.")
      else:
          if self.is_local:
              if self.env:
                  os.environ.update(self.env)
              import ray
  
              kwargs = self._update_extra_params(self.extra_params)
              init_params = dict(
                  num_cpus=self.ray_node_cpu_cores,
                  object_store_memory=self.object_store_memory,
                  include_dashboard=self.include_webui,
                  dashboard_host="0.0.0.0",
              )
              init_params.update(kwargs)
              if self.redis_password:
                  init_params["_redis_password"] = self.redis_password
              if version.parse(ray.__version__) >= version.parse(
                  "1.4.0"
              ):
                  init_params["namespace"] = "az"
              self._address_info = ray.init(**init_params)
          else:
              self.cluster_ips = self._gather_cluster_ips()
              redis_address = self._start_cluster()
              self._address_info = self._start_driver(
                  num_cores=driver_cores, redis_address=redis_address
              )
  
          print(self._address_info)
          kill_redundant_log_monitors(
              self._address_info["redis_address"]
          )
          self.initialized = True
      return self._address_info
  
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 8 for calling the RayContext obj's underlying member method code with example usage: obj.init(driver_cores) and returns: self