&`iz,(:ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddlm Z ddl m Z mZmZmZddlZddlZddlmZddlZddlZddlmZddlmZddlmZmZmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+dd l,m-Z-m.Z.dd l/m0Z0dd l1m2Z2dd l3m4Z4m5Z5ej6d Z7e78ej9dZ:dZ;dZGddZ?dZ@dZAdZBdZCeDejEFddZGdZHeDejEFddZIdZJdZKd ZLd!ZMd"ZNd#ZOd$eDd%eDd&eDd'eDd(eDd)eDd*ePd+eDd,eDd-eDd.eDd/e d0e d1eQd2eQd3eRd4eRd5ePd6ee?f&d7ZSdaTejUZVd8ZWd1ddddddddd9 ZXd1dddd&d(dd-dddd: ZYd;ZZd$eDd%eeDd&eeDd'eeDd(eeDd)eeDd+eeDd,eeDd-eeDd.eeDd/ee d0ee d1eeQddd?d?d@d$eDd%eeDd&eeDd'eeDd(eeDd)eeDdAeeDdBeeDd-eeDd.eeDd/ee d0ee d1eeQddd?d?dEd$eDdFePd%eeDd&eeDd'eeDd(eeDd)eeDdAeeDdBeeDd-eeDd.eeDd/ee d0ee dacdMZddS)ON)Event)DictOptionalTupleType)Version)$DefaultDatabricksRayOnSparkStartHook)RayOnSparkStartHook)_get_cpu_cores_get_local_ray_node_slots_get_num_physical_gpus_wait_service_upcalc_mem_ray_head_nodeexec_cmdgen_cmd_exec_failure_msg!get_avail_mem_per_ray_worker_node*get_configured_spark_executor_memory_bytesget_max_num_concurrent_tasksget_random_unused_port!get_spark_application_driver_hostget_spark_session%get_spark_task_assigned_physical_gpusis_in_databricks_runtimeis_port_in_use) build_address parse_address) load_class) HEAD_NODE_ID) DeveloperAPI PublicAPIzray.util.sparkRAY_ON_SPARK_START_HOOK RAY_ON_SPARK_COLLECT_LOG_TO_PATH!RAY_ON_SPARK_START_RAY_PARENT_PIDctjdkrtdd} ddl}t |jjdkrt|dS#t$rt|wxYw)Nposixz3Ray on spark only supports running on POSIX system.z-ray.util.spark module requires pyspark >= 3.3r)r(r)osname RuntimeErrorpysparkr __version__release ImportError)spark_dependency_errorr,s o/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/spark/cluster_init.py_check_system_environmentr2:s w'PQQQL3 7& ' ' /) ; ;566 6 < ; 33312223s 0AA/c<eZdZdZdZdZdZdZdZdZ dZ d S) RayClusterOnSparkaW This class is the type of instance returned by the `_setup_ray_cluster` interface. Its main functionality is to: Connect to, disconnect from, and shutdown the Ray cluster running on Apache Spark. Serve as a Python context manager for the `RayClusterOnSpark` instance. Args address: The url for the ray head node (defined as the hostname and unused port on Spark driver node) head_proc: Ray head process spark_job_group_id: The Spark job id for a submitted ray job num_workers_node: The number of workers in the ray cluster. c ||_||_||_||_||_||_||_||_| |_| |_ | |_ d|_ d|_ d|_ d|_dS)NF)address head_procmin_worker_nodesmax_worker_nodestemp_dircluster_unique_id start_hookray_dashboard_portspark_job_serverglobal_cluster_lock_fdray_client_server_port is_shutdownspark_job_is_canceledbackground_job_exceptionray_ctx) selfr6r7r8r9r:r;r<r=r>r?r@s r1__init__zRayClusterOnSpark.__init__Ws} " 0 0  !2$"4 0&<#&<# %*"(,% cddl}|jrtd |j|j|jSt t|jd|jtr |j |jnS tdnB#t$r5}tdt|dYd}~nd}~wwxYwd}t!j} t!jt$|jtd|jt)d |jDd z }||jkr0td |d  |jdS||kr=|}t!j}td |d|jdn t!j|z t4kr|dkr|jjdd\}}t;jdt?||dddi}| |j!"d} tGj$| } | d} | rtd| tdtd|d|jd |jdS#|jwxYw)Nrz9The ray cluster has been shut down or it failed to start.r6zray.dashboard.optional_depszDependencies to launch the optional dashboard API server cannot be found. They can be installed with pip install ray[default], root cause: ()TzRay workers failed to start.c"g|] }|d | S)Alive).0nodes r1 z6RayClusterOnSpark.wait_until_ready..s!GGG$gGGGGrGr zStarted zH Ray worker nodes, meet the minimum number of Ray worker nodes required.z*Ray worker nodes are starting. Progress: (z / http://z/query_last_worker_errspark_job_group_idurljsonzutf-8last_worker_errz(Starting Ray worker node failed, error: zBCurrent spark cluster has no resources to launch Ray worker nodes.zQTimeout in waiting for minimal ray workers to start. Started / Total requested: (zs). Current spark cluster does not have sufficient resources to launch requested minimal number of Ray worker nodes.)%rayrAr+initr6r=rr_RAY_DASHBOARD_STARTUP_TIMEOUTr<on_ray_dashboard_created __import__ModuleNotFoundError_loggerwarningreprtimesleep/_RAY_CLUSTER_STARTUP_PROGRESS_CHECKING_INTERVALrClennodesr8infoshutdownr9*_RAY_CONNECT_CLUSTER_POLL_PROGRESS_TIMEOUTr>server_addressrequestspostrraise_for_statuscontentdecoderVloads) rErXelast_alive_worker_countlast_progress_move_timecur_alive_worker_countjob_server_hostjob_server_portresponse decoded_respjson_resrWs r1wait_until_readyz"RayClusterOnSpark.wait_until_readyxs   K ]  CHT\ * * * *&27Gdl++A.'.882 889PQQQQ<====*OOMBFq''MMM'( #&*ikk #F  JKKK 0<&689 GG)#)++GGGHH1L'*T-BBBLLQ#9QQQf CLNNNNNc*,CCC.D+.2ikk+LLP2PP7;7LPPP  &==DEE2Q66!% 5 DRaR H / /'/}%=mO_.].]%=%=%=';D%A (((H%55777+3+;+B+B7+K+KL'+z,'?'?H.67H.IO. "&2%9'6%9%9'"'"!" '3%8'"'"!" V 6VV;?;PVVV CLNNNNNQF P CLNNNNsDA0K BK C'+CKCB KDrgr7 terminater^r_r`)rErrps r1rgzRayClusterOnSpark.shutdowns3   $ !!!!     JNN=$ / / /*6 D7GGG  ! * * , , , ((****   WdSTggWW  $D   ' $ $s$" //B-- C+7*C&&C+c|SrrMr|s r1 __enter__zRayClusterOnSpark.__enter__s rGc.|dSr)rg)rEexc_typeexc_valexc_tbs r1__exit__zRayClusterOnSpark.__exit__s rGN) __name__ __module__ __qualname____doc__rFryr}rrgrrrMrGr1r4r4Hs  BeeeN666 $$$6rGr4cd|dd}|dvr|dtj|S||S|dt|S)N--_-) system_config resourceslabels=)replacerVdumpsstr)keyvalue converted_keys r1_convert_ray_node_optionr sl0S#..00M 66655$*U"3"3555 } * *c%jj * **rGc>d|DS)Nc4g|]\}}t||SrM)r)rNkvs r1rPz-_convert_ray_node_options..s' G G Gtq! $Q * * G G GrG)items)optionss r1_convert_ray_node_optionsrs G Gw}} G G GGrG<(RAY_ON_SPARK_BACKGROUND_JOB_STARTUP_WAIT30r(-RAY_ON_SPARK_RAY_WORKER_NODE_STARTUP_INTERVAL10xc2ddlddlfd}d} ||nA#t$r4 tj|n#t $rYnwxYw||YnwxYwfd d}tj|r{t|d5}| }dddn #1swxYwYd | d D}d |D}nAg}t|d 5 dddn #1swxYwYtj |d fd|D}t|}t|dkrd}n,t!|dz}t#|D] } | |vr| }n ||tj<t|d 5}|D] \} } || d| d ! dddn #1swxYwYd|dzz} | dz} | dkrt+dn#t $r wxYwfd}t-j|d| | fS)ad If we start multiple ray workers on a machine concurrently, some ray worker processes might fail due to ray port conflicts, this is because race condition on getting free port and opening the free port. To address the issue, this function use an exclusive file lock to delay the worker processes to ensure that port acquisition does not create a resource contention issue due to a race condition. After acquiring lock, it will allocate port range for worker ports (for ray node config --min-worker-port and --max-worker-port). Because on a spark cluster, multiple ray cluster might be created, so on one spark worker machine, there might be multiple ray worker nodes running, these worker nodes might belong to different ray cluster, and we must ensure these ray nodes on the same machine using non-overlapping worker port range, to achieve this, in this function, it creates a file `/tmp/ray_on_spark_worker_port_allocation.txt` file, the file format is composed of multiple lines, each line contains 2 number: `pid` and `port_range_slot_index`, each port range slot allocates 1000 ports, and corresponding port range is: - range_begin (inclusive): 20000 + port_range_slot_index * 1000 - range_end (exclusive): range_begin + 1000 In this function, it first scans `/tmp/ray_on_spark_worker_port_allocation.txt` file, removing lines that containing dead process pid, then find the first unused port_range_slot_index, then regenerate this file, and return the allocated port range. Returns: Allocated port range for current worker ports rNctjtjztjz} tj||}tj|dd}t |D]N} |jj z|cS#t$rYnwxYwtj dOtd|d#t$rtj|YdSwxYw)NiX zAcquiring lock on file z timeout.)r)O_RDWRO_CREATO_TRUNCopenchmodrangerLOCK_EXLOCK_NBBlockingIOErrorrarb TimeoutErrorrclose) file_pathmodefd max_lock_iterrrs r1 acquire_lockz8_preallocate_ray_worker_port_range..acquire_lockCsy2:% 2 D))B HY ' ' 'M=))  KKEMEM$ABBBIII 'D 2MMMMNN N    HRLLLLLL s5=C'#B C BCB+CC('C(z2/tmp/ray_on_spark_worker_startup_barrier_lock.lockcfjtjdSr)rrr)r)rlock_fdsr1 release_lockz8_preallocate_ray_worker_port_range..release_lockhs/ GU]+++ rGz,/tmp/ray_on_spark_worker_port_allocation.txtr)rc8g|]}|dS) )split)rNlines r1rPz6_preallocate_ray_worker_port_range..ts/   $( 3   rG cPg|]#\}}t|t|f$SrM)int)rNpid_strslot_index_strs r1rPz6_preallocate_ray_worker_port_range..ws?   +G^Ws>223   rGwrcFi|]\}}|||SrM) pid_exists)rNpid slot_indexpsutils r1 z6_preallocate_ray_worker_port_range..sC   Z  %%    rGr r NiizqToo many ray worker nodes are running on this machine, cannot allocate worker port range for new ray worker node.cNtjtdSr)rarb!_RAY_WORKER_NODE_STARTUP_INTERVAL)rsr1 hold_lockz5_preallocate_ray_worker_port_range..hold_locks# 4555 rGrM)targetargs)rrrr)removerpathexistsrreadstriprrsetvaluesrdmaxrgetpidrwriter+ threadingThreadstart)rlock_file_pathport_alloc_filefpport_alloc_dataport_alloc_tableport_alloc_mapallocated_slot_setnew_slot_indexindexrrworker_port_range_beginworker_port_range_endrrrrrs @@@@r1"_preallocate_ray_worker_port_ranger#si8LLLMMM.JN /,~..  / / /   In % % % %    D ,~.. /7H 7>>/ * * .oC000 ,B"$'')) , , , , , , , , , , , , , , ,  ,;,A,A,C,C,I,I$,O,O      /?      " oC000                  H_f - - -    #3    !!6!6!8!899 ! " "a ' 'NN !344q8N~..   222%*NE3'5ry{{# / , , , 2#1#7#7#9#9 2 2ZC00*0001111 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2#(.4*?"? 7$ > 5 ( (F  )   IB///55777 "$9 99s ! AAA A A AAA)2IB<0 I<CICAID& I&D**I-D*.B0I6H! I!H%%I(H%)%II%chd|vri|d<|d}d|vrtjdd|id|d<|S)Nrobject_spilling_config filesystemdirectory_path)typeparams)rVr)head_node_optionsobject_spilling_dirsys_confs r1#_append_default_spilling_dir_configrse///-//* 1Hx//-1Z$$&9  . . )* rGcNd|vri|d<|d||S)Nr)update) node_optionsrs r1_append_resources_configrs6,&&$& [!$$Y/// rGc~tjtjdddS)N RAY_TMPDIRz/tmprX)r)rjoinrgetrMrGr1_get_default_ray_tmp_dirr s( 7<< |V<z"*%<=>>@@@ ! # #.3I>>>"9---rGr9r8num_cpus_worker_nodenum_cpus_head_nodenum_gpus_worker_nodenum_gpus_head_nodeusing_stage_schedulingheap_memory_worker_nodeheap_memory_head_nodeobject_store_memory_worker_nodeobject_store_memory_head_noderworker_node_optionsray_temp_root_dircollect_log_to_pathautoscale_upscaling_speedautoscale_idle_timeout_minutesrreturnc  ddl}t|}t}tjt |}t |dd}|g}| } | dd}| dd}|rd}nt |dd| }| |t |dd| }| |||d urh|(t |dd| }| |t |dd| }| |d d |d |g}|d ur|dgz }ndg}t d|d|d|dtj jdd}|r| tdt!jdt jt jzt jz} |||j|jzn#t0$rtdwxYwd} t3}!t!j|!d t j|!d}"ntd}| |} t j| d|d|} t!j| d t j| d}"t!j|"d t=| |"} ddlm }#|!}$|#||||$}%tE|||| dd|||| di||di||||| | ||d ||!}&|&#|||| || ||$\}'}(|&j$})tKj&tNtQ||sd|')(|'*tKj&d"tW|)|'j,|(}*t[d#|*zt d$t]||}+|+t j/d%<ta|+|'||| ||||%||& },|1|,|,S)'a The public API `ray.util.spark.setup_ray_cluster` does some argument validation and then pass validated arguments to this interface. and it returns a `RayClusterOnSpark` instance. The returned instance can be used to connect to, disconnect from and shutdown the ray cluster. This instance can also be used as a context manager (used by encapsulating operations within `with _setup_ray_cluster(...):`). Upon entering the managed scope, the ray cluster is initiated and connected to. When exiting the scope, the ray cluster is disconnected and shut down. Note: This function interface is stable and can be used for instrumentation logging patching. rNi(#i'min_portmax_portinclude_dashboarddashboard_porti')r!r" exclude_listTz--dashboard-host=0.0.0.0z--dashboard-port=--dashboard-agent-listen-port=z--include-dashboard=truez--include-dashboard=falsezRay head hostname: z, port: z, ray client server port: .zXRay on spark global mode cluster does not allow you to set 'ray_temp_root_dir' argument.z%/tmp/ray_on_spark_global_cluster.lockzAcquiring global lock failed for setting up new global mode Ray on spark cluster. If there is an active global mode Ray on spark cluster, please shut down it before you create a new one.)exist_okspillzray-r)_start_spark_job_server)CPUGPUmemoryobject_store_memoryz ray.worker)r node_config min_workers max_workers) ray_head_ip ray_head_portr;r ray_temp_dirrrspark_job_server_port)head_resourcesworker_node_typesextra_provider_configupscaling_speedidle_timeout_minutesg?zStart Ray head node failed! zRay head node started.r) r6r7r8r9r:r;r<r=r>r?r@)2rrrsocket gethostbynamerrcopyrappendr^rfuuiduuid4hex ValueErrorr)rrrrrrrrr makedirsrr get_default_temp_root_dirr.ray.autoscaler._private.spark.spark_job_serverr+custom_environment_variablesAutoscalingClusterrray_head_node_cmdrarb_RAY_HEAD_STARTUP_TIMEOUTrpollrr returncoder+rrr4on_cluster_created)-r9r8rrrrrrrrrrrrrrrrrr<sparkr3r4port_exclude_listr#r=r@r6ray_dashboard_agent_portdashboard_optionsr;r?r5ray_default_tmp_dirrr+ray_node_custom_envr>autoscaling_cluster ray_head_proctail_output_dequerIcmd_exec_failure_msgcluster_addressray_cluster_handlers- r1_setup_ray_clusterrZsFLLL#I..J   E&'H'O'OPPK*;PUVVVM&*..00)--.A4HH*../?FF !&!7 * " " " 34442&  2333 $5$=$=  %!7. """   $ $%7 8 8 8#9 * $ $ $    !9::: ' 4 2 4 4 G-E G G   $ $ "HH!%  $ * D D F F w|| ImII6GII   L40000 gll<AAK#d3333;.%AACC..*E3F-%%+#@    //5+J  "//    '*!2&<(#6#6%:   2;?   B(;'@'@ ( ($M$,= J())) +} 5 5 S      '  # # % % % JsOOO7 }79J  :=QQRRR LL)***#K??O /BJ}+))+-)55   !!"5666 s /#HH-cddlm}ddlm}||}|dkr|d|}||jS)Nr)ResourceProfileBuilder)TaskResourceRequestsgpu)pyspark.resource.profiler\pyspark.resource.requestsr]cpusresourcerequirebuild)num_cpus_per_nodenum_gpus_per_noder\r] task_res_reqs r1_create_resource_profilerhs??????>>>>>>''))../@AAL1#,,U4EFF ! ! # # + +L 9 9 ??rG) r:blockheadnode_ip_addressportnum_cpusnum_gpusdashboard_hostdashboard_agent_listen_port) r:rirjr6rmrnr.r/rpmin_worker_portmax_worker_portc|D]h}|dsd|vrtd||vr:d|d|d}||}|rt|d|dt|d idS) NrrzFor a ray node option like '--foo-bar', you should convert it to following format 'foo_bar' in 'head_node_options' / 'worker_node_options' arguments.zSetting the option 'z' for z nodes is not allowed.z You should set the 'z' option instead.z+ This option is controlled by Ray on Spark.) startswithrC)r block_keys node_typercommon_err_msgreplacement_args r1_verify_node_optionsrys >>$   3#::3  *  SsSS)SSS )oO  %O !%RRR rG strict_modec tt| pi} | pi} t| tdt| tdt t dtjrt dt}|j j }|dkp| d}| ds,| ds|d ks|st d |rd }n_tr+ttjd jd krd}n&ddl}t|jjdkrd}nd }d|vr3|t+d|d}t-jdt0d|vr3|t+d|d}t-jdt0d|vr3|t+d|d}t-jdt0t3|j dd}||dkrt+dt9|j dd}t3|}|dkrLd|d }trdd!lm}|d"|d#nt> |||dkrt+d$d%}|j !d&gd&"|#d\}}} |-|+|rd}!tI||}"nKt+d'|d(|d)|$|"|r|}|}d}!tI||}"nd }!d}"|}|}nt+d*tK|||||\}#}$tM||||}%tO|}&|&|%|#|$zzz}'|'d+| zkrHd,}trdd!lm}|d-|d.nt> |d/|vrt+d0|tPkr)|t+d1tS|j |"}|}n|dkrt+d2d3|vrt+d4||}nd|cxkr|ksnt+d5g}(|d6kr|(*d7|d8|#d9kr|(*d:|#d;|(rQ| r"t+d<+|(t> d=+|(||rtY}nd}n|dkrt+d>|d?|'|r" t[}n.#t\$rd}Yn wxYwd}n|dkrt+d@|d?|dkr |dkr| dA}dA} nt_|| \}} t`5tcdVidB|dC|dD|dE|dF|dG|dH|!dI|#dJ|dK|$dL| dM| dN| dO| dP|dQ|dR|dS|})|)a |)2nB#t\$r5}* tgn#t\$rYnwxYwt dT|*d}*~*wwxYw dddn #1swxYwYti|)j5d}+dUtm|+|)j7},|)j5|,fS)WNzRay head node on sparkzRay worker node on sparkzCurrent active ray cluster on spark haven't shut down. Please call `ray.util.spark.shutdown_ray_cluster()` before initiating a new Ray cluster on spark.zCurrent python process already initialized Ray, Please shut down it by `ray.shutdown()` before initiating a Ray cluster on spark.localzlocal[zspark://zlocal-cluster[yarnzxRay on Spark only supports spark cluster in standalone mode, local-cluster mode, spark on yarn mode or spark local mode.FDATABRICKS_RUNTIME_VERSION Tr)r(rrezi'num_cpus_per_node' and 'num_cpus_worker_node' arguments are equivalent. Only set 'num_cpus_worker_node'.z_'num_cpus_per_node' argument is deprecated, please use 'num_cpus_worker_node' argument instead.rfzi'num_gpus_per_node' and 'num_gpus_worker_node' arguments are equivalent. Only set 'num_gpus_worker_node'.z_'num_gpus_per_node' argument is deprecated, please use 'num_gpus_worker_node' argument instead.object_store_memory_per_nodez'object_store_memory_per_node' and 'object_store_memory_worker_node' arguments are equivalent. Only set 'object_store_memory_worker_node'.zu'object_store_memory_per_node' argument is deprecated, please use 'object_store_memory_worker_node' argument instead.zspark.task.cpus1z2Argument `num_cpus_worker_node` value must be > 0.zspark.task.resource.gpu.amount0z3You configured 'spark.task.resource.gpu.amount' to z,we recommend setting this value to 0 so that Spark jobs do not reserve GPU resources, preventing Ray-on-Spark workloads from having the maximum number of GPUs available.)$get_databricks_display_html_functionzzz3Argument `num_gpus_worker_node` value must be >= 0.c\ddlm}m}m}|}|}|}|||fS)Nr)r r'_get_spark_worker_total_physical_memory)ray.util.spark.utilsr rr)rr rrnum_cpus_spark_workernum_gpus_spark_workertotal_mem_bytess r1_get_spark_worker_resourcesz@_setup_ray_cluster_internal.._get_spark_worker_resourcessv          !/ 0 0 6 6 8 8AACC " !   rGr a Current spark version does not support stage scheduling, so that you cannot set the argument `num_cpus_worker_node` and `num_gpus_worker_node` values. Without setting the 2 arguments, per-Ray worker node will be assigned with number of 'spark.task.cpus' (equals to zS) cpu cores and rounded down number of 'spark.task.resource.gpu.amount' (equals to z) GPUs. To enable spark stage scheduling, you need to upgrade spark to 3.4 version or use Databricks Runtime 12.x, and you cannot use spark local mode.zb'num_cpus_worker_node' and 'num_gpus_worker_node' arguments must beset together or unset together.g?aDIn each spark worker node, we recommend making the sum of 'spark_executor_memory + num_Ray_worker_nodes_per_spark_worker * (memory_worker_node + object_store_memory_worker_node)' to be less than 'spark_worker_physical_memory * 0.8', otherwise it might lead to spark worker physical memory exhaustion and Ray task OOM errors.z"z
num_worker_nodeszn'num_worker_nodes' argument is removed, please set 'max_worker_nodes' and 'min_worker_nodes' argument instead.zIf you set 'max_worker_nodes' to 'MAX_NUM_WORKER_NODES', autoscaling is not supported, so that you cannot set 'min_worker_nodes' argument and 'min_worker_nodes' is automatically set to be equal to 'max_worker_nodes'.ztThe value of 'max_worker_nodes' argument must be either a positive integer or 'ray.util.spark.MAX_NUM_WORKER_NODES'. autoscalez'autoscale' argument is removed. You can set 'min_worker_nodes' argument to be less than 'max_worker_nodes' to make autoscaling enabled.zZThe value of 'max_worker_nodes' argument must be an integer >= 0 and <= 'max_worker_nodes'rzThe provided CPU resources for each ray worker are inadequate to start a ray cluster. Based on the total cpu resources available and the configured task sizing, each ray worker node would start with aF CPU cores. This is less than the recommended value of `4` CPUs per worker. On spark version >= 3.4 or Databricks Runtime 12.x, you can set the argument `num_cpus_worker_node` to a value >= 4 to address it, otherwise you need to increase the spark application configuration 'spark.task.cpus' to a minimum of `4` to address it.l zThe provided memory resources for each ray worker node are inadequate. Based on the total memory available on the spark cluster and the configured task sizing, each ray worker would start with a bytes heap memory. This is less than the recommended value of 10GB. The ray worker node heap memory size is calculated by (SPARK_WORKER_PHYSICAL_MEMORY / num_local_spark_task_slots * 0.8) - object_store_memory_worker_node. To increase the heap space available, increase the memory in the spark cluster by using instance types with larger memory, or increase number of CPU/GPU per Ray worker node (so it leads to less Ray worker node slots per spark worker node), or apply a lower `object_store_memory_worker_node`.zYou are creating ray cluster on spark with strict mode (it can be disabled by setting argument 'strict_mode=False' when calling API 'setup_ray_cluster'), strict mode requires the spark cluster config satisfying following criterion: rzCArgument `num_cpus_head_node` value must be >= 0. Current value is r'zBArgument `num_gpus_head_node` value must be >= 0.Current value is i@r9r8rrrrrrrrrrrrrrrrz"Launch Ray-on-Spark cluster failedzray://rM)8r2_install_sigterm_signalry_head_node_option_block_keys_worker_node_option_block_keys_active_ray_clusterr+rXr{r sparkContextmasterrtrrr)rmajorr,r-r.rCwarningswarnDeprecationWarningrgetConfr floatray.util.spark.databricks_hookrr^r_ parallelizemapcollectrhrr rMAX_NUM_WORKER_NODESrr?r r rrr_active_ray_cluster_rwlockrZryshutdown_ray_clusterrr6rr@)-r9r8rrrrrrrrrrrrzrrrrkwargsrN spark_masteris_spark_local_modesupport_stage_schedulingr,num_spark_task_cpusdefault_num_spark_task_gpusrounded_num_spark_task_gpuswarn_msgrrrrspark_worker_mem_bytesr res_profileray_worker_node_heap_mem_bytes&ray_worker_node_object_store_mem_bytesspark_worker_ray_node_slotsspark_executor_memory_bytes"spark_worker_required_memory_bytesinsufficient_resourcesclusterrphead_ipremote_connection_addresss- r1_setup_ray_cluster_internalrs .)/R-3$  &" &       L     E%,L&'1V\5L5LX5V5V  ++   " "#3 4 4   6 ! !  " J    -#(   "" - BJ;< = = Cr I I#'   7& ' ' /9 < <'+ $ $', $f$$  +?  &&9:  7    f$$  +?  &&9:  7    &// * 65  +11O*P'  B    e088::>>?PRUVVWW',@A,E,EMNNN#( ""$$(()I3OO###&&A"B"B"Q&& 0* 0 0 0  $ % % &       3 0 0 2 27777     OOH % % %',@1,D,DNOOO   $ &&sA..  ( ) ) 1 L13I ',@,L # %) "2$&:KKP1D PP : PPP     %*>*F # ?$9 #8 %) "2$&:KK&+ "K#6 #>  .    * '   &.#< ###MU"S"S# % ),R R T T' *C2H,HHH O  $ % % &       3 0 0 2 2GXGGG     OOH % % %V## J   ///  '& 88JKXX+ Q   @   f N   +#7777'77777 (    a%% $    &(???%% B. B B B  ?  ?T011   OODII&<== > > >!  #!/!1!1  !"    ! !:$6:::  !  # '%;%=%="" ' ' '%&""" '"#    ! !:$6:::  a ! # # ) 1 !3(:%%?U !#@@ @ << $ L L$   -- -- "6!5  21  "6!5  21  $:#9 %C$B #8"7 -S,R +H*G 0/ !4 3 0/ !4 3 '@&?! ",J+I# $ i% ,& L  $ $ & & & & L L L $&&&&    CDD! K  L '5 L L L L L L L L L L L L L L LDGO,,Q/GIw(FGGII ?5 55sg)V88 WWAZ;Y,+Z;, Z+7ZZ& ZZ&ZZ&&Z++Z;;Z?Z?Fg?)r8rrrrmemory_worker_nodememory_head_noderrrrrrzrrrrrc tdid|d|d|d|d|d|d|d|d |d | d | d | d | d| d|d|d|dd|S)a 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, 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: 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: 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: 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: 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: 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: 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 ":" "remote_connection_address" is in format of "ray://:", 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)``. r9r8rrrrrrrrrrrrzrrrrFrM)r)r9r8rrrrrrrrrrrrzrrrrs r1setup_ray_clusterrs ` '   )) )) 21 .-   21   .-  !3 2 /. )H(G 'D&C ,+ 0/ ,+  K 0/  #<";! "(F'E# $% '  rGT) is_blockingr8rrrrrrrrrrrzrrrrc(tdid|d|d|d|d|d|d|d|d | d | d | d | d dd| d|d|d|dd}|s|S tatdat dS#dat wxYw)a1 Set up a global mode cluster. The global Ray on spark cluster means: - You can only create one active global Ray on spark cluster at a time. On databricks cluster, the global Ray cluster can be used by all users, - as contrast, non-global Ray cluster can only be used by current notebook user. - It is up persistently without automatic shutdown. - On databricks notebook, you can connect to the global cluster by calling ``ray.init()`` without specifying its address, it will discover the global cluster automatically if it is up. For global mode, the ``ray_temp_root_dir`` argument is not supported. Global model Ray cluster always use the default Ray temporary directory path. All arguments are the same with ``setup_ray_cluster`` API except that: - the ``ray_temp_root_dir`` argument is not supported. Global model Ray cluster always use the default Ray temporary directory path. - A new argument "is_blocking" (default ``True``) is added. If "is_blocking" is True, then keep the call blocking until it is interrupted. once the call is interrupted, the global Ray on spark cluster is shut down and `setup_global_ray_cluster` call terminates. If "is_blocking" is False, once Ray cluster setup completes, return immediately. r9r8rrrrrrrrrrrNrzrrrrTrM)rr _global_ray_cluster_cancel_eventwaitr)r9rr8rrrrrrrrrrrzrrrrXs r1setup_global_ray_clusterrhsgd2))))21.-  21  .- !3 2/.)H(G'D&C,+0/$ K0/ #<";!"(F'E#$$%O*  +077((--///+/(  ,0( s 'A??Bc * |j}|jd|j   f d}|j||jt t||}|r%t }| |}tdu}| | | d\}}|s ||_|SdS)Nr c 3 Kddlm}tjd}|}t \}}t dd}tjddd d d td d d|d|dz d|g t}| dtpdtttjddi}dkr|} d| vrt%dd| djD} t)| } | dt+| dd| D|d<|dd| d} t1jd td!d"#} | d$sd%} t%d&t1jd td'd(i#t7|d%|)dS#t8$r1}d*|d+}t:||| fVYd}~dSd}~wwxYw),Nr) TaskContextzray.util.spark.workeri'rr -mray.util.spark.start_ray_nodez --num-cpus=--blockz --address=z --memory=z--object-store-memory=--min-worker-port=--max-worker-port=r r& --temp-dir=!RAY_ENABLE_WINDOWS_OR_OSX_CLUSTERrr^zDCouldn't get the gpu id, Please check the GPU resource configurationcPg|]#}t|$SrM)rr)rNaddrs r1rPzK_start_ray_worker_nodes..ray_cluster_job_mapper..s5&*DJJLL!!rGz --num-gpus=,c,g|]}t|SrM)r)rNgpu_ids r1rPzK_start_ray_worker_nodes..ray_cluster_job_mapper.."sCCCVCCCrGCUDA_VISIBLE_DEVICESzStart Ray worker, command: rFrRz/check_node_id_availability)node_idrSrT availableTzDStarting Ray worker node twice with the same node id is not allowed.z/notify_task_launchedrS synchronous extra_envz&Ray worker node process exit, reason: r')pyspark.taskcontextrlogging getLoggerr rrsys executablerrr?r$r%rr)rrr+ addressesrrdr rfrjrkrVrrr^r_)rr_worker_loggercontextrr$ray_worker_node_dashboard_agent_portray_worker_node_cmdray_worker_node_extra_envstask_resources gpu_addr_listavailable_physical_gpusis_task_reschedule_failurervrperr_msgrheap_memory_per_noderrerfrr3r4rSr5rSr6rs r1ray_cluster_job_mapperz7_start_ray_worker_nodes..ray_cluster_job_mappers333333 *+BCC//## / 0 0 # !0F %%0 0 0 , N  + -+ - -  D{MBB D D ., . . C%A C C :!8 : :   I#((3F*G*G I I   E 6). &}2mK9NOO222 '*<   H==??;/ .2*"& M,mK9NOO,,,)*<      # 4        6 6 6 DqCCCG OOG $ $ $55 5 5 5 5 5 5 5 5 5' 6s5BI J&JJr r)rNrirSr setJobGrouprlistrrh withResourcesron_spark_job_created mapPartitionsrlast_worker_error)r>rSspark_job_group_descrrr3r4r5rerfrrrrrrNrjob_rddresource_profile hook_entryrrrSr6s ` `````````` @@r1_start_ray_worker_nodesrsH  "E,;A>*>G6G6G6G6G6G6G6G6G6G6G6G6G6G6G6G6G6R ""0 ,, U# $ $%%'7G:3    ''(899#|t/CEEEJ##$6777*1*?*?++ gii+'G ' &-4* 4rGct5ttdtdaddddS#1swxYwYdS)z+ Shut down the active ray cluster. Nz#No active ray cluster to shut down.)rrr+rgrMrGr1rrs $##  &DEE E$$&&&" ##################s2AA A c :eZdZdZdededededef dZdZd Zd S) rHz*Create a ray on spark autoscaling cluster.r7r8r9r:r;c||_t|jd<|||||||_dS)zCreate the cluster. Args: head_resources: resources of the head node, including CPU. worker_node_types: autoscaler node types config for worker nodes. NODE_ID_AS_RESOURCEN)r>_head_resourcesr_generate_config_config)rEr7r8r9r:r;s r1rFzAutoscalingCluster.__init__sP .22446B23,,   !     rGc tjttjtjtjd}tj |}||d<|idd|dd<td| D|d<|d |||d <||d <|S) Nzautoscaler/spark/defaults.yamlavailable_node_typesr)rr0r2zray.head.defaultc3,K|]\}}|dVdS)r2NrM)rNrrs r1 z6AutoscalingCluster._generate_config..s<+ + !%AAm + + + + + + rGr2providerr:r;)yaml safe_loadrr)rr dirnamerX__file__r>deepcopysumrr)rEr7r8r9r:r; base_config custom_configs r1rz#AutoscalingCluster._generate_configsn  GOOCL114     k22 0A ,-'E E ,-.@A (++ + ):)@)@)B)B+ + + ( ( m$ j!(()>???+: '(0D ,-rGc Dddlm} m} m} |!tj|d} n,tjtd} t| d5} | tj |j dddn #1swxYwYt\}}tjdddd d |d |d |d | d|d|dz g |}||d|d|jvr@|d|jdd|jvr@|d|jdd|jvr@|d|jdd|jvr@|d|jd| ||j}|| |dd| |pdt*t-t ji|}||_t3|d|S)z{Start the cluster. After this call returns, you can connect to the cluster with ray.init("auto"). r)r$rrNzautoscaling_config.jsonrrrrz--headz--node-ip-address=z--port=z--ray-client-server-port=z--autoscaling-config=rrr rr,z --num-cpus={}r-z --num-gpus={}r.z --memory={}r/z--object-store-memory={}AUTOSCALER_UPDATE_INTERVAL_SrrFr)ray.util.spark.cluster_initr$rrr)rr r rrrVrrrrrr?rformatrextendr%rrrI_start_ray_head_node)rEr3r4r@r5rQrrrSr$rrautoscale_configfrrrIrs r1rzAutoscalingCluster.startsp            #!w||L:STT  !w||(**,E   "C ( ( .A GGDJt|,, - - - . . . . . . . . . . . . . . . / 0 0 # ! N  +   . . . %m % % @(> @ @ 6$4 6 6 :!8 : : .preexec_functionNs rG)rr preexec_fn)r)rIrrrs r1r r Ms6    #    rGctrdS tjtjfd}tjtj|dadS#t$rt dYdSwxYw)Nc tn#t$rYnwxYwtjtjt jt jtjdSr)rrsignalSIGTERMr)killr)signumframe_origin_sigterm_handlers r1_sigterm_handlerz1_install_sigterm_signal.._sigterm_handlerqs $&&&&     M 7    G V^     s  Tz,Install Ray-on-Spark SIGTERM handler failed.)_sigterm_signal_installedr getsignalrrr^r_)rrs @r1rrhs!H"("26>"B"B       fn&6777$(!!! HHHFGGGGGGHsAA$A:9A:)rN)er>rVrr)rr<rrrar@rrtypingrrrrrjrpackaging.versionrrXray._private.servicesdatabricks_hookr start_hook_baser utilsr r rrrrrrrrrrrrrrray._common.network_utilsrrray._common.utilsr+ray.autoscaler._private.spark.node_providerrray.util.annotationsr r!rr^setLevelINFOr"rr$r%r2r4rrrJrZrrr _BACKGROUND_JOB_STARTUP_WAITrcrrhrrrr rboolrrrZrRLockrrhrrryrrrrrrrHr rrrMrGr1r-s    ............ %%%%%% AAAAAA000000$CBBBBBBB((((((DDDDDD88888888 ' , - -4#E $G! 3 3 3D+++HHH!#"sJNN=tDD  34/$'CJNNBDII%%!.1*H:H:H:V    EEE...www w  w  ww!w!ww&)w$'wwww !w" %#w$%*%w&'w( )wwwwt,Y_..@@@$   #'  $  &&<#' " "2n6n6smn6#3-n6! n6 #3- n6 ! n6&c]n6$C=n6&.c]n6$,C=n6 ~n6"$n6 }n6n6"#n6 (!n6"%-UO#n6$%n6( 38_)n6n6n6n6b  '+*.(,*.(,(,&*5937(,*.'+)-1469%ccccsmc#3- c ! c #3- c! c! csmc&.c]c$,C=c ~c"$c }cc "#!c" (#c$%-UO%c( 38_)ccc cL &**.(,*.(,(,&*5937(,*.)-1469%SSSSSsm S #3- S ! S#3-S! S! SsmS&.c]S$,C=S ~S"$SS "#!S" (#S$%-UO%SSS Sl^^^B  # # #  #$( W W W W W W W W t0"HHHHHrG