o `۷i3 @sddlZddlmZddlmZmZmZmZmZddl Z ddl m Z ddl Z ddl mZddlmZddlmZddlmZdd lmZmZdd lmZdd lmZeeZd Z d e!de"de#de$de!f ddZ%eddGdddeZdS)N)partial)AnyCallableDictOptionalUnion)Version) Checkpoint)TRAIN_DATASET_KEY) GenDataset)_log_deprecation_warning)RayTrainReportCallback XGBoostConfig)XGBoostTrainer) PublicAPIaKPassing in `xgboost.train` kwargs such as `params`, `num_boost_round`, `label_column`, etc. to `XGBoostTrainer` is deprecated in favor of the new API which accepts a training function, similar to the other DataParallelTrainer APIs (ex: TorchTrainer). See this issue for more context: https://github.com/ray-project/ray/issues/50042config label_columnnum_boost_round dataset_keysxgboost_train_kwargscCstj}d}|}|r$t|}|}||}td|d|dtjt } | } dd|D} dd| D} | j |dd| |} }tj| |d }|t fg}| D]\}}|j |dd||}}|tj||d |fq\i}tj|f|||||d |dS) Nz7Model loaded from checkpoint will train for additional zY iterations (trees) in order to achieve the target number of iterations (num_boost_round=z).cSs"i|] }|tkr|tj|qS)r raytrainget_dataset_shard).0krrW/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/train/xgboost/xgboost_trainer.py 6s  z0_xgboost_train_fn_per_worker..cSsi|] \}}||qSr) materialize to_pandas)rrdrrrr;s)axis)label)dtrainevals evals_resultr xgb_model)rrget_checkpointr get_modelnum_boosted_roundsloggerinforr rritemsdropxgboostDMatrixappend)rrrrr checkpointstarting_modelremaining_iters starting_iter train_ds_itertrain_df eval_ds_iterseval_dfstrain_Xtrain_yr$r% eval_nameeval_dfeval_Xeval_yr&rrr_xgboost_train_fn_per_workersL      r@beta) stabilitycsFeZdZdZdZdZ ddddddddddddd deeegdfee gdffdee dee dee j j d ee j jd ee eefd ee j jd eed ee eefdeedee eefdeeffddZde d ee j jd ee eefdeedeedee gdff ddZededejfddZZS)raA Trainer for distributed data-parallel XGBoost training. Example ------- .. testcode:: :skipif: True import xgboost import ray.data import ray.train from ray.train.xgboost import RayTrainReportCallback, XGBoostTrainer def train_fn_per_worker(config: dict): # (Optional) Add logic to resume training state from a checkpoint. # ray.train.get_checkpoint() # 1. Get the dataset shard for the worker and convert to a `xgboost.DMatrix` train_ds_iter, eval_ds_iter = ( ray.train.get_dataset_shard("train"), ray.train.get_dataset_shard("validation"), ) train_ds, eval_ds = train_ds_iter.materialize(), eval_ds_iter.materialize() train_df, eval_df = train_ds.to_pandas(), eval_ds.to_pandas() train_X, train_y = train_df.drop("y", axis=1), train_df["y"] eval_X, eval_y = eval_df.drop("y", axis=1), eval_df["y"] dtrain = xgboost.DMatrix(train_X, label=train_y) deval = xgboost.DMatrix(eval_X, label=eval_y) params = { "tree_method": "approx", "objective": "reg:squarederror", "eta": 1e-4, "subsample": 0.5, "max_depth": 2, } # 2. Do distributed data-parallel training. # Ray Train sets up the necessary coordinator processes and # environment variables for your workers to communicate with each other. bst = xgboost.train( params, dtrain=dtrain, evals=[(deval, "validation")], num_boost_round=10, callbacks=[RayTrainReportCallback()], ) train_ds = ray.data.from_items([{"x": x, "y": x + 1} for x in range(32)]) eval_ds = ray.data.from_items([{"x": x, "y": x + 1} for x in range(16)]) trainer = XGBoostTrainer( train_fn_per_worker, datasets={"train": train_ds, "validation": eval_ds}, scaling_config=ray.train.ScalingConfig(num_workers=4), ) result = trainer.fit() booster = RayTrainReportCallback.get_model(result.checkpoint) Args: train_loop_per_worker: The training function to execute on each worker. This function can either take in zero arguments or a single ``Dict`` argument which is set by defining ``train_loop_config``. Within this function you can use any of the :ref:`Ray Train Loop utilities `. train_loop_config: A configuration ``Dict`` to pass in as an argument to ``train_loop_per_worker``. This is typically used for specifying hyperparameters. xgboost_config: The configuration for setting up the distributed xgboost backend. Defaults to using the "rabit" backend. See :class:`~ray.train.xgboost.XGBoostConfig` for more info. datasets: The Ray Datasets to use for training and validation. dataset_config: The configuration for ingesting the input ``datasets``. By default, all the Ray Datasets are split equally across workers. See :class:`~ray.train.DataConfig` for more details. scaling_config: The configuration for how to scale data parallel training. ``num_workers`` determines how many Python processes are used for training, and ``use_gpu`` determines whether or not each process should use GPUs. See :class:`~ray.train.ScalingConfig` for more info. run_config: The configuration for the execution of the training run. See :class:`~ray.train.RunConfig` for more info. resume_from_checkpoint: A checkpoint to resume training from. This checkpoint can be accessed from within ``train_loop_per_worker`` by calling ``ray.train.get_checkpoint()``. metadata: Dict that should be made available via `ray.train.get_context().get_metadata()` and in `checkpoint.get_metadata()` for checkpoints saved from this Trainer. Must be JSON-serializable. label_column: [Deprecated] Name of the label column. A column with this name must be present in the training dataset. params: [Deprecated] XGBoost training parameters. Refer to `XGBoost documentation `_ for a list of possible parameters. num_boost_round: [Deprecated] Target number of boosting iterations (trees in the model). Note that unlike in ``xgboost.train``, this is the target number of trees, meaning that if you set ``num_boost_round=10`` and pass a model that has already been trained for 5 iterations, it will be trained for 5 iterations more, instead of 10 more. **train_kwargs: [Deprecated] Additional kwargs passed to ``xgboost.train()`` function. TN) train_loop_configxgboost_configscaling_config run_configdatasetsdataset_configresume_from_checkpointmetadatarparamsrtrain_loop_per_workerrCrDrErFrGrHrIrJrrKrc  s|ttjtdkr td|du}|r"|j| || | |d}| p i}n | r+tdttt|j ||||||||| d dS)Nz1.7.0zm`XGBoostTrainer` requires the `xgboost` version to be >= 1.7.0. Upgrade with: `pip install -U "xgboost>=1.7"`)rrFrrrGzPassing `xgboost.train` kwargs to `XGBoostTrainer` is deprecated. In your training function, you can call `xgboost.train(**kwargs)` with arbitrary arguments. ) rLrCrDrErFrGrHrIrJ) rr/ __version__ ImportError_get_legacy_train_fn_per_workerr *LEGACY_XGBOOST_TRAINER_DEPRECATION_MESSAGEsuperr__init__)selfrLrCrDrErFrGrHrIrJrrKr train_kwargs legacy_api __class__rrrRs>   zXGBoostTrainer.__init__rreturnc Cs|pi}|tstdtdt||std|p d}tt|dg}tdd|D}i}|rN|jj } |jj } | |d<| d urJ| nd |d <|sZ| t d i|||d<t t||t||d } | S)zs  zAXGBoostTrainer._get_legacy_train_fn_per_worker.. frequencyNTcheckpoint_at_end)rrrrr)getr ValueErrorlistkeysr rPanycheckpoint_configcheckpoint_frequencyr`r1r rr@set) rSrrFrGrrrZuser_supplied_callbackcallback_kwargsrgr`train_fn_per_workerrrrrOsJ   z.XGBoostTrainer._get_legacy_train_fn_per_workerr2cCs t|S)z5Retrieve the XGBoost model stored in this checkpoint.)r r))clsr2rrrr)3s zXGBoostTrainer.get_modelr[)__name__ __module__ __qualname____doc___handles_checkpoint_freq_handles_checkpoint_at_endrrrrrrr ScalingConfig RunConfigstrr DataConfigr rintrRrO classmethodr/Boosterr) __classcell__rrrVrrTs~f      9   ;r)&logging functoolsrtypingrrrrrr/packaging.versionr ray.trainrr ray.train.constantsr ray.train.trainerr ray.train.utilsr ray.train.xgboostr rray.train.xgboost.v2rSimpleXGBoostTrainerray.util.annotationsr getLoggerrmr+rPdictrurwrhr@rrrrs:           6