code: def setup_ray_cluster(
      *,
      max_worker_nodes: int,
      min_worker_nodes: Optional[int] = None,
      num_cpus_worker_node: Optional[int] = None,
      num_cpus_head_node: Optional[int] = None,
      num_gpus_worker_node: Optional[int] = None,
      num_gpus_head_node: Optional[int] = None,
      object_store_memory_worker_node: Optional[int] = None,
      object_store_memory_head_node: Optional[int] = None,
      head_node_options: Optional[Dict] = None,
      worker_node_options: Optional[Dict] = None,
      ray_temp_root_dir: Optional[str] = None,
      strict_mode: bool = False,
      collect_log_to_path: Optional[str] = None,
      autoscale_upscaling_speed: Optional[float] = 1.0,
      autoscale_idle_timeout_minutes: Optional[float] = 1.0,
      **kwargs,
  ) -> Tuple[str, str]:
      """
      Set up a ray cluster on the spark cluster by starting a ray head node in the
      spark application's driver side node.
      After creating the head node, a background spark job is created that
      generates an instance of `RayClusterOnSpark` that contains configuration for the
      ray cluster that will run on the Spark cluster's worker nodes.
      After a ray cluster is set up, "RAY_ADDRESS" environment variable is set to
      the cluster address, so you can call `ray.init()` without specifying ray cluster
      address to connect to the cluster. To shut down the cluster you can call
      `ray.util.spark.shutdown_ray_cluster()`.
      Note: If the active ray cluster haven't shut down, you cannot create a new ray
      cluster.
  
      Args:
          max_worker_nodes: This argument represents maximum ray worker nodes to start
              for the ray cluster. you can
              specify the `max_worker_nodes` as `ray.util.spark.MAX_NUM_WORKER_NODES`
              represents a ray cluster
              configuration that will use all available resources configured for the
              spark application.
              To create a spark application that is intended to exclusively run a
              shared ray cluster in non-scaling, it is recommended to set this argument
              to `ray.util.spark.MAX_NUM_WORKER_NODES`.
          min_worker_nodes: Minimal number of worker nodes (default `None`),
              if "max_worker_nodes" value is equal to "min_worker_nodes" argument,
              or "min_worker_nodes" argument value is None, then autoscaling is disabled
              and Ray cluster is launched with fixed number "max_worker_nodes" of
              Ray worker nodes, otherwise autoscaling is enabled.
          num_cpus_worker_node: Number of cpus available to per-ray worker node, if not
              provided, use spark application configuration 'spark.task.cpus' instead.
              **Limitation** Only spark version >= 3.4 or Databricks Runtime 12.x
              supports setting this argument.
          num_cpus_head_node: Number of cpus available to Ray head node, if not provide,
              use 0 instead. Number 0 means tasks requiring CPU resources are not
              scheduled to Ray head node.
          num_gpus_worker_node: Number of gpus available to per-ray worker node, if not
              provided, use spark application configuration
              'spark.task.resource.gpu.amount' instead.
              This argument is only available on spark cluster that is configured with
              'gpu' resources.
              **Limitation** Only spark version >= 3.4 or Databricks Runtime 12.x
              supports setting this argument.
          num_gpus_head_node: Number of gpus available to Ray head node, if not provide,
              use 0 instead.
              This argument is only available on spark cluster which spark driver node
              has GPUs.
          object_store_memory_worker_node: Object store memory available to per-ray worker
              node, but it is capped by
              "dev_shm_available_size * 0.8 / num_tasks_per_spark_worker".
              The default value equals to
              "0.3 * spark_worker_physical_memory * 0.8 / num_tasks_per_spark_worker".
          object_store_memory_head_node: Object store memory available to Ray head
              node, but it is capped by "dev_shm_available_size * 0.8".
              The default value equals to
              "0.3 * spark_driver_physical_memory * 0.8".
          head_node_options: A dict representing Ray head node extra options, these
              options will be passed to `ray start` script. Note you need to convert
              `ray start` options key from `--foo-bar` format to `foo_bar` format.
              For flag options (e.g. '--disable-usage-stats'), you should set the value
              to None in the option dict, like `{"disable_usage_stats": None}`.
              Note: Short name options (e.g. '-v') are not supported.
          worker_node_options: A dict representing Ray worker node extra options,
              these options will be passed to `ray start` script. Note you need to
              convert `ray start` options key from `--foo-bar` format to `foo_bar`
              format.
              For flag options (e.g. '--disable-usage-stats'), you should set the value
              to None in the option dict, like `{"disable_usage_stats": None}`.
              Note: Short name options (e.g. '-v') are not supported.
          ray_temp_root_dir: A local disk path to store the ray temporary data. The
              created cluster will create a subdirectory
              "ray-{head_port}-{random_suffix}" beneath this path.
          strict_mode: Boolean flag to fast-fail initialization of the ray cluster if
              the available spark cluster does not have sufficient resources to fulfill
              the resource allocation for memory, cpu and gpu. When set to true, if the
              requested resources are not available for recommended minimum recommended
              functionality, an exception will be raised that details the inadequate
              spark cluster configuration settings. If overridden as `False`,
              a warning is raised.
          collect_log_to_path: If specified, after ray head / worker nodes terminated,
              collect their logs to the specified path. On Databricks Runtime, we
              recommend you to specify a local path starts with '/dbfs/', because the
              path mounts with a centralized storage device and stored data is persisted
              after Databricks spark cluster terminated.
          autoscale_upscaling_speed: If autoscale enabled, it represents the number of
              nodes allowed to be pending as a multiple of the current number of nodes.
              The higher the value, the more aggressive upscaling will be. For example,
              if this is set to 1.0, the cluster can grow in size by at most 100% at any
              time, so if the cluster currently has 20 nodes, at most 20 pending launches
              are allowed. The minimum number of pending launches is 5 regardless of
              this setting.
              Default value is 1.0, minimum value is 1.0
          autoscale_idle_timeout_minutes: If autoscale enabled, it represents the number
              of minutes that need to pass before an idle worker node is removed by the
              autoscaler. The smaller the value, the more aggressive downscaling will be.
              Worker nodes are considered idle when they hold no active tasks, actors,
              or referenced objects (either in-memory or spilled to disk). This parameter
              does not affect the head node.
              Default value is 1.0, minimum value is 0
      Returns:
          returns a tuple of (address, remote_connection_address)
          "address" is in format of "<ray_head_node_ip>:<port>"
          "remote_connection_address" is in format of
          "ray://<ray_head_node_ip>:<ray-client-server-port>",
          if your client runs on a machine that also hosts a Ray cluster node locally,
          you can connect to the Ray cluster via ``ray.init(address)``,
          otherwise you can connect to the Ray cluster via
          ``ray.init(remote_connection_address)``.
      """
  
      return _setup_ray_cluster_internal(
          max_worker_nodes=max_worker_nodes,
          min_worker_nodes=min_worker_nodes,
          num_cpus_worker_node=num_cpus_worker_node,
          num_cpus_head_node=num_cpus_head_node,
          num_gpus_worker_node=num_gpus_worker_node,
          num_gpus_head_node=num_gpus_head_node,
          object_store_memory_worker_node=object_store_memory_worker_node,
          object_store_memory_head_node=object_store_memory_head_node,
          head_node_options=head_node_options,
          worker_node_options=worker_node_options,
          ray_temp_root_dir=ray_temp_root_dir,
          strict_mode=strict_mode,
          collect_log_to_path=collect_log_to_path,
          autoscale_upscaling_speed=autoscale_upscaling_speed,
          autoscale_idle_timeout_minutes=autoscale_idle_timeout_minutes,
          is_global=False,
          **kwargs,
      )
  
imports: import copy
  import yaml
  import json
  import os
  import socket
  import sys
  import time
  import threading
  import logging
  import uuid
  import warnings
  import requests
  from packaging.version import Version
  from typing import Optional, Dict, Tuple, Type
  import ray
  import ray._private.services
  from ray.autoscaler._private.spark.node_provider import HEAD_NODE_ID
  from ray.util.annotations import DeveloperAPI, PublicAPI
  from ray._private.storage import _load_class
  from .utils import exec_cmd, is_port_in_use, get_random_unused_port, get_spark_session, get_spark_application_driver_host, is_in_databricks_runtime, get_spark_task_assigned_physical_gpus, get_avail_mem_per_ray_worker_node, get_max_num_concurrent_tasks, gen_cmd_exec_failure_msg, calc_mem_ray_head_node, _wait_service_up
  from .start_hook_base import RayOnSparkStartHook
  from .databricks_hook import DefaultDatabricksRayOnSparkStartHook
  from threading import Event
label_desc: Use image node_id 13 for calling a global function with example usage: setup_ray_cluster() and returns: _setup_ray_cluster_internal