o ciBB@s*ddlZddlZddlZddlZddlZddlZddlZddlZddlZe dZ ddZ ddZ ddZ dd d d d Zd dZddZ d9ddZddZddZddZddZddZd Zd Zd!d"Zd#d$Zd%d&Zd'Zd(Zd)Zd*Zd+Z d,Z!d-d.Z"d/d0Z#d1d2Z$d3d4Z%d5d6Z&d7d8Z'dS):Nzray.util.spark.utilscCs dtjvS)NDATABRICKS_RUNTIME_VERSION)osenvironrrH/home/ubuntu/.local/lib/python3.10/site-packages/ray/util/spark/utils.pyis_in_databricks_runtimes rcCs,d|}d|}d|d|d|dS)N zCommand z failed with return code z", tail output are included below.  )join)cmd return_codetail_output_dequecmd_str tail_outputrrrgen_cmd_exec_failure_msgs  rcCsD|jdd}t|dd}|d}ddddd}|||S) Nzspark.executor.memory1gii@l)kmgt)confgetlowerint)spark value_str value_num value_unitunit_maprrr*get_configured_spark_executor_memory_bytess r"T) extra_env synchronousc st|hd}|rtdt||dd}|dur(|dur(td|dur.|nitj|}tj |f|dtj tj d|t j dd fd d }tj|d d |safS}|dkrqtt||dS)a A convenience wrapper of `subprocess.Popen` for running a command from a Python script. If `synchronous` is True, wait until the process terminated and if subprocess return code is not 0, raise error containing last 100 lines output. If `synchronous` is False, return an `Popen` instance and a deque instance holding tail outputs. The subprocess stdout / stderr output will be streamly redirected to current process stdout. >textstderrstdoutz`kwargs` cannot contain envNz5`extra_env` and `env` cannot be used at the same timeT)r(r%r'r&d)maxlencs&jD] }|tj|qdSN)r'appendsyswrite)lineprocessrrrredirect_log_thread_fnOs  z(exec_cmd..redirect_log_thread_fnr)targetargsr)setkeys intersection ValueErrorlistpoprr subprocessPopenPIPESTDOUT collectionsdeque threadingThreadstartwait RuntimeErrorr)r r#r$kwargsillegal_kwargsr(r2r rr0rexec_cmd)s8   rHcCs^ddl}ddlm}|||j|j}|||fdkWdS1s(wYdS)Nr)closing)socket contextlibrIAF_INET SOCK_STREAM connect_ex)hostportrJrIsockrrris_port_in_usecs  $rRcCsDt}t||kr t||rdStdt||ks dS)NTF)timerRsleep)rOrPtimeoutbeg_timerrr_wait_service_upks  rXrr)cCs\t}|pg}t|D]}|||}||vrq t||s"|Sq td|d|d)z! Get random unused port. z!Get available port between range z and z failed.)random SystemRandomrangerandintrRrE)rOmin_portmax_port max_retries exclude_listrng_rPrrrget_random_unused_portvs   rdcCs(ddlm}|}|durtd|S)Nr) SparkSessionzSpark session haven't been initiated yet. Please use `SparkSession.builder` to create a spark session and connect to a spark cluster.) pyspark.sqlregetActiveSessionrE)re spark_sessionrrrget_spark_sessions ricCs |jdS)Nzspark.driver.host)rr)rrrr!get_spark_application_driver_hosts rjcCsV|j}|dur"dd}|dgd||||jS|| S)z0Gets the current max number of concurrent tasks.NcSsdSr+r)rcrrr dummpy_mappersz3get_max_num_concurrent_tasks..dummpy_mapperrS) _jscsc parallelize withResourcesmapcollectmaxNumConcurrentTasks_java_resource_profileresourceProfileManagerdefaultResourceProfile) spark_contextresource_profilesscrkrrrget_max_num_concurrent_taskss   rycCs*ddl}ttjvrttjtS|jSNr)psutil)RAY_ON_SPARK_WORKER_PHYSICAL_MEMORY_BYTESrrrvirtual_memorytotal)r{rrr'_get_spark_worker_total_physical_memorys  rcCs,ddl}ttjvrttjtS|djSNrz/dev/shm)shutil'RAY_ON_SPARK_WORKER_SHARED_MEMORY_BYTESrrr disk_usager~)rrrr%_get_spark_worker_total_shared_memorys  rg?c Csddl}ddl}ttjvrttjt}n|j}|t}t tjvr+ttjt }n| dj}|t}t ||||\}}}|durHt |||fSr)r{r)RAY_ON_SPARK_DRIVER_PHYSICAL_MEMORY_BYTESrrrr}r~'_RAY_ON_SPARK_NODE_MEMORY_BUFFER_OFFSET'RAY_ON_SPARK_DRIVER_SHARED_MEMORY_BYTESr_calc_mem_per_ray_node_loggerwarning) configured_heap_memory_bytesconfigured_object_store_bytesr{ravailable_physical_memavailable_shared_memheap_mem_bytesobject_store_bytes warning_msgrrrcalc_mem_ray_head_nodes.      rcCs.t||t}t||t}t||||Sr+)rrr)num_task_slotsphysical_mem_bytesshared_mem_bytesrravailable_physical_mem_per_nodeavailable_shared_mem_per_noderrr_calc_mem_per_ray_worker_nodes  rc Csddlm}m}d}|p||}tjds||kr|}|t}||kr(|}d}||kr?||kr7d|d}nd|d}|}t|}|durNt||} nt|} | ||fS) Nr)&DEFAULT_OBJECT_STORE_MEMORY_PROPORTION!OBJECT_STORE_MINIMUM_MEMORY_BYTES#RAY_OBJECT_STORE_ALLOW_SLOW_STORAGEzzYour configured `object_store_memory_per_node` value is too high and it is capped by 80% of per-Ray node allocated memory.zYour operating system is configured with too small /dev/shm size, so `object_store_memory_worker_node` value is configured to minimal size (z- bytes),Please increase system /dev/shm size.zYou configured too small Ray node object store memory size, so `object_store_memory_worker_node` value is configured to minimal size (zI bytes),Please increase 'object_store_memory_worker_node' argument value.)ray._private.ray_constantsrrrrr0_RAY_ON_SPARK_MAX_OBJECT_STORE_MEMORY_PROPORTIONr) rrrrrrrrobject_store_bytes_upper_boundrrrrrs@  rRAY_ON_SPARK_WORKER_CPU_CORESRAY_ON_SPARK_WORKER_GPU_NUMr|rrrcCs(ddl}ttjvrttjtS|Srz)multiprocessingrrrr cpu_count)rrrr_get_cpu_coresSs rc Csttjvr ttjtStddurdSztjdddddd}t|j dWSt yG}zt dt|WYd}~dSd}~ww)Nz nvidia-smirz1nvidia-smi --query-gpu=name --format=csv,noheaderT)shellcheckr%capture_outputr zU'nvidia-smi --query-gpu=name --format=csv,noheader' command execution failed, error: )rrrrrwhichr;runlenr'stripsplit Exceptionrinforepr)completed_procerrr_get_num_physical_gpusas, rcCsd||krtd|d|d||}|dkr0||kr&td|d|d|||kr0||}|S)Nzscpu number per Ray worker node should be <= spark worker node CPU cores, you set cpu number per Ray worker node to z* but spark worker node CPU core number is .rz|gpu number per Ray worker node should be <= spark worker node GPU number, you set GPU devices number per Ray worker node to z- but spark worker node GPU devices number is )r8)num_cpusnum_gpusnum_cpus_per_nodenum_gpus_per_nodenum_ray_node_slotsrrr_get_local_ray_node_slots~s* rc CsVt}|dkr t}nd}t||||}t}t}t|||||\} } } | | d| fS)z Returns tuple of ( ray_worker_node_heap_mem_bytes, ray_worker_node_object_store_bytes, error_message, # always None warning_message, ) rN)rrrrrr) rrheap_memory_per_nodeobject_store_memory_per_noderrrrrray_worker_node_heap_mem_bytes"ray_worker_node_object_store_bytesrrrr"_get_avail_mem_per_ray_worker_nodes2 rc sjfdd}|jdgd|d\}}}} |dur(td|d| dur1t| ||fS)a< Return the available heap memory and object store memory for each ray worker, and error / warning message if it has. Return value is a tuple of (ray_worker_node_heap_mem_bytes, ray_worker_node_object_store_bytes, error_message, warning_message) NB: We have one ray node per spark task. c sdztWSty1}zddl}d||j}ddt||dfWYd}~Sd}~ww)Nrr r)rr tracebackr format_tb __traceback__r)rcrr trace_msgrrrrrrmappers z1get_avail_mem_per_ray_worker_node..mapperrSrNz:Inferring ray worker node available memory failed, error: a. You can bypass this error by setting following spark configs: spark.executorEnv.RAY_ON_SPARK_WORKER_CPU_CORES, spark.executorEnv.RAY_ON_SPARK_WORKER_GPU_NUM, spark.executorEnv.RAY_ON_SPARK_WORKER_PHYSICAL_MEMORY_BYTES, spark.executorEnv.RAY_ON_SPARK_WORKER_SHARED_MEMORY_BYTES.) sparkContextrnrprqrErr) rrrrrr'inferred_ray_worker_node_heap_mem_bytes+inferred_ray_worker_node_object_store_byteserrrrrr!get_avail_mem_per_ray_worker_nodes   rcs:dtjvrddtjddDfdd|DS|S)NCUDA_VISIBLE_DEVICEScSsg|]}t|qSr)rr).0devrrr s z9get_spark_task_assigned_physical_gpus..,csg|]}|qSrr)raddrvisible_cuda_dev_listrrrs)rrr) gpu_addr_listrrr%get_spark_task_assigned_physical_gpuss r)rrYr)N)(r;rr-rZrAr?loggingrrT getLoggerrrrr"rHrRrXrdrirjryrrrrrrrrrr|rrrrrrrrrrrrrsT   :   %E. ;