o ڭj @sddZddlZddlZddlZddlZddlZddlmZddlm Z m Z m Z m Z m Z mZmZddlZddlmZmZmZmZmZddlmZddlmZdd lmZdd lmZdd lmZdd l m!Z!dd l"m#Z#ddl$m%Z%dede&fddZ'de de e&de e&e ffddZ(GdddeZdDde&de)de)defddZ*dede)defdd Z+d!ede&fd"d#Z,defd$d%Z-dEd&e&d'e ee&e)fdej.fd(d)Z/d&e&de e)fd*d+Z0d,ede)fd-d.Z1d,ede2fd/d0Z3dede2fd1d2Z4d3ede)fd4d5Z5de&fd6d7Z6d8e&d9e ge#fde#fd:d;Z7dZ8d8e&de!fd?d@Z9dAe e&de2fdBdCZ:dS)Fz;Xgboost pyspark integration submodule for helper functions.N)Thread)AnyCallableDictOptionalSetTypeUnion)BarrierTaskContext SparkConf SparkContext SparkFiles TaskContext) SparkSession)CommunicatorContext)Config)_Args)_ArgVals)Booster)XGBModel) RabitTrackerclsreturncCs|jd|jS)zReturn the class name..) __module____name__)rrP/var/www/html/CropPilot/venv/lib/python3.10/site-packages/xgboost/spark/utils.pyget_class_namesrfuncunsupported_setcCsDt|}i}|jD]}|j|jur|j|vr|j||j<q |S)zReturns a dictionary of parameters and their default value of function fn. Only the parameters with a default value will be included. )inspect signature parametersvaluesdefaultemptyname)r r!sigfiltered_params_dict parameterrrr_get_default_params_from_funcs    r,cs.eZdZdZdededdffdd ZZS)rz&Context with PySpark specific task ID.contextargsrNc s&t||d<tjdi|dS)N dmlc_task_idr)str partitionIdsuper__init__)selfr-r. __class__rrr35szCommunicatorContext.__init__)rr __qualname____doc__r CollArgsValsr3 __classcell__rrr5rr2s"rhost n_workersportcCsLd|i}t||d|d}|t|jd}d|_||||S)z"Start Rabit tracker with n_workersr<task)r<host_ipsortbyr=)targetT)rstartrwait_fordaemonupdate worker_args)r;r<r=r.trackerthreadrrr_start_tracker:s rIconfcCs4|jdusJ|jdurdn|j}t|j||}|S)z3Get rabit context arguments to send to each worker.Nr)tracker_host_ip tracker_portrI)rJr<r=envrrr_get_rabit_argsFsrNr-cCsdd|D}|dS)zLGets the hostIP for Spark. This essentially gets the IP of the first worker.cSsg|] }|jddqS):r)addresssplit).0inforrr Psz _get_host_ip..r) getTaskInfos)r- task_ip_listrrr _get_host_ipNsrWcCs tjdur tdtjS)z`Get or create spark session. Note: This function can only be invoked from driver side. Nz<_get_spark_session should not be invoked from executor side.)pysparkrget RuntimeErrorrbuilder getOrCreaterrrr_get_spark_sessionTs  r]r(levelcCstt|}|dur||n |jtjkr|tj|js8tjs8ttj }t d}| || ||S)zGGets a logger by name, or creates and configures it for the first time.Nz<%(asctime)s %(levelname)s %(name)s: %(funcName)s %(message)s) logging getLoggersetLevelr^NOTSETINFOhandlers StreamHandlersysstderr Formatter setFormatter addHandler)r(r^loggerhandler formatterrrr get_loggeras       rncCs t|}|jtjkr dS|jS)z+Get the logger level for the given log nameN)r_r`r^rb)r(rkrrrget_logger_levelus ro spark_contextcCs@|jdkr|j|jdS|jS)z0Gets the current max number of concurrent tasks.z3.1r)_jscscversionmaxNumConcurrentTasksresourceProfileManagerresourceProfileFromIdrprrr_get_max_num_concurrent_tasks{s  rxcCs|jS)zWhether it is Spark local mode)rqrrisLocalrwrrr _is_localsrzcCs&|d}|duo|dp|dS)Nz spark.masterzspark://z local-cluster)rY startswith)rJmasterrrr_is_standalone_or_localclusters r} task_contextcCs>|durtd|}d|vrtdt|djdS)z&Get the gpu id from the task resourcesNz3_get_gpu_id should not be invoked from driver side.gpuzDCouldn't get the gpu id, Please check the GPU resource configurationr)rZ resourcesint addressesstrip)r~rrrr _get_gpu_idsrcCs0t}tj|d}tj|st||S)Nz xgboost-tmp)r getRootDirectoryospathjoinexistsmakedirs)root_dir xgb_tmp_dirrrr_get_or_create_tmp_dirs   rmodelxgb_model_creatorcCs|}|t|d|S)zH Deserialize an xgboost.XGBModel instance from the input model. utf-8) load_model bytearrayencode)rr xgb_modelrrrdeserialize_xgb_modelsrboostercCs^tjttd}||t|dd }|}Wd|S1s(wY|S)z Serialize the input booster to a string. Parameters ---------- booster: an xgboost.core.Booster instance .jsonrencodingN) rrrruuiduuid4 save_modelopenread)r tmp_file_namefser_model_stringrrrserialize_boosters   rcCsft}tjttd}t|ddd }||Wdn1s'wY| ||S)zN Deserialize an xgboost.core.Booster from the input ser_model_string. rwrrN) rrrrrrrrwriter)rrrrrrrdeserialize_boosters  rdevicecCs|dvS)z&Whether xgboost is using CUDA workers.)cudarr)rrrruse_cudasr)r)N);r8r"r_rrfr threadingrtypingrrrrrrr rXr r r r rpyspark.sql.sessionr collectiverCCtxrrCollArgsrr9corersklearnrrGrr0rr,rrIrNrWr]Loggerrnrorxboolrzr}rrrrrrrrrrs` $           &