setup_ray_cluster#

ray.util.spark.setup_ray_cluster(*, max_worker_nodes: int, min_worker_nodes: int | None = None, num_cpus_worker_node: int | None = None, num_cpus_head_node: int | None = None, num_gpus_worker_node: int | None = None, num_gpus_head_node: int | None = None, memory_worker_node: int | None = None, memory_head_node: int | None = None, object_store_memory_worker_node: int | None = None, object_store_memory_head_node: int | None = None, head_node_options: Dict | None = None, worker_node_options: Dict | None = None, ray_temp_root_dir: str | None = None, strict_mode: bool = False, collect_log_to_path: str | None = None, autoscale_upscaling_speed: float | None = 1.0, autoscale_idle_timeout_minutes: float | None = 1.0, **kwargs) → Tuple[str, str][source]#

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. Ray on Spark uses Autoscaler V1 by default. Set the environment variable RAY_enable_autoscaler_v2=1 before calling this function to enable Autoscaler V2.

Parameters:

max_worker_nodes (int) – 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 (int | None) – 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 (int | None) – Number of cpus available to per-ray worker node, if not provided, if spark stage scheduling is supported, ‘num_cpus_head_node’ value equals to number of cpu cores per spark worker node, otherwise it uses 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 (int | None) – Number of cpus available to Ray head node, if not provide, if it is global mode Ray cluster, use number of cpu cores in spark driver node, otherwise use 0 instead. use 0 instead. Number 0 means tasks requiring CPU resources are not scheduled to Ray head node.
num_gpus_worker_node (int | None) – Number of gpus available to per-ray worker node, if not provided, if spark stage scheduling is supported, ‘num_gpus_worker_node’ value equals to number of GPUs per spark worker node, otherwise it uses rounded down value of 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 (int | None) – Number of gpus available to Ray head node, if not provide, if it is global mode Ray cluster, use number of GPUs in spark driver node, otherwise use 0 instead. This argument is only available on spark cluster which spark driver node has GPUs.
memory_worker_node (int | None) – Optional[int]: Heap memory configured for Ray worker node. This is basically setting --memory option when starting Ray node by ray start command.
memory_head_node (int | None) – Optional[int]: Heap memory configured for Ray head node. This is basically setting --memory option when starting Ray node by ray start command.
object_store_memory_worker_node (int | None) – 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 (int | None) – 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 (Dict | None) – 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 (Dict | None) – 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 (str | None) – 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 (bool) – 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 (str | None) – 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 (float | None) – 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 (float | None) – 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
**kwargs – Additional keyword arguments forwarded to _setup_ray_cluster_internal for advanced/experimental options.

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 type:

Tuple[str, str]