o `۷i4 @sddlZddlmZddlmZmZmZmZmZddl Z ddl Z ddl m Z ddl mZddlmZddlmZddlmZdd lmZdd lmZdd lmZeeZd Zd e de!de"de#de f ddZ$eddGdddeZdS)N)partial)AnyCallableDictOptionalUnion) Checkpoint)TRAIN_DATASET_KEY)RayTrainReportCallback)LightGBMConfig)LightGBMTrainer) GenDataset)_log_deprecation_warning) PublicAPIa\Passing in `lightgbm.train` kwargs such as `params`, `num_boost_round`, `label_column`, etc. to `LightGBMTrainer` is deprecated in favor of the new API which accepts a `train_loop_per_worker` argument, 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_keyslightgbm_train_kwargsc CsFtj}d}|}|r$t|}|}||}td|d|dtjt } | } dd|D} dd| D} | j |dd| |} }tj| |d }|g}t g}| D]\}}|j |dd||}}|tj||d ||q]|tjj|d d |d d tjd||||||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).0krrY/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/train/lightgbm/lightgbm_trainer.py 6s  z1_lightgbm_train_fn_per_worker..cSsi|] \}}||qSr) materialize to_pandas)rrdrrrr;s)axis)label tree_learner data_parallel pre_partitionT)params train_setr valid_sets valid_names init_modelr)rrget_checkpointr get_modelcurrent_iterationloggerinforr rritemsdroplightgbmDatasetappendupdateget_network_params setdefault)rrrrr checkpointstarting_modelremaining_iters starting_iter train_ds_itertrain_df eval_ds_iterseval_dfstrain_Xtrain_yr'r(r) eval_nameeval_dfeval_Xeval_yrrr_lightgbm_train_fn_per_workersR        rFbeta) 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)r aA Trainer for distributed data-parallel LightGBM training. Example ------- .. testcode:: :skipif: True import lightgbm import ray.data import ray.train from ray.train.lightgbm import RayTrainReportCallback, LightGBMTrainer 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 `lightgbm.Dataset` 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 = lightgbm.Dataset(train_X, label=train_y) deval = lightgbm.Dataset(eval_X, label=eval_y) params = { "objective": "regression", "metric": "l2", "learning_rate": 1e-4, "subsample": 0.5, "max_depth": 2, # Adding the line below is the only change needed # for your `lgb.train` call! **ray.train.lightgbm.get_network_params(), } # 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 = lightgbm.train( params, train_set=dtrain, valid_sets=[deval], valid_names=["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 = LightGBMTrainer( 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. lightgbm_config: The configuration for setting up the distributed lightgbm backend. Defaults to using the "rabit" backend. See :class:`~ray.train.lightgbm.LightGBMConfig` 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] LightGBM training parameters. Refer to `LightGBM documentation `_ for a list of possible parameters. num_boost_round: [Deprecated] Target number of boosting iterations (trees in the model). Note that unlike in ``lightgbm.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 ``lightgbm.train()`` function. TN) train_loop_configlightgbm_configscaling_config run_configdatasetsdataset_configresume_from_checkpointmetadatarr&rtrain_loop_per_workerrIrJrKrLrMrNrOrPrr&rc  sl|du}|r|j| || | |d}| pi}n| r#td| dttt|j||||||||| d dS)N)rrLrrrMzPPassing `lightgbm.train` kwargs to `LightGBMTrainer` is deprecated. Got kwargs: z^ In your training function, you can call `lightgbm.train(**kwargs)` with arbitrary arguments. ) rQrIrJrKrLrMrNrOrP)_get_legacy_train_fn_per_workerrkeys+LEGACY_LIGHTGBM_TRAINER_DEPRECATION_MESSAGEsuperr __init__)selfrQrIrJrKrLrMrNrOrPrr&r train_kwargs legacy_api __class__rrrVs:   zLightGBMTrainer.__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)z=Get the training function for the legacy LightGBMTrainer API.z`datasets` must be provided for the LightGBMTrainer API if `train_loop_per_worker` is not provided. This dict must contain the training dataset under the key: 'z '. Got keys: z`label_column` must be provided for the LightGBMTrainer API if `train_loop_per_worker` is not provided. This is the column name of the label in the dataset. callbackscss|]}t|tVqdSN) isinstancer )rcallbackrrr s  zBLightGBMTrainer._get_legacy_train_fn_per_worker.. frequencyNTcheckpoint_at_end)rrrrr)getr ValueErrorlistrSrrTanycheckpoint_configcheckpoint_frequencyrdr4r rrFset) rWrrLrMrrr^user_supplied_callbackcallback_kwargsrjrdtrain_fn_per_workerrrrrRsJ   z/LightGBMTrainer._get_legacy_train_fn_per_workerr8cCs t|S)z6Retrieve the LightGBM model stored in this checkpoint.)r r,)clsr8rrrr,6s zLightGBMTrainer.get_modelr_)__name__ __module__ __qualname____doc___handles_checkpoint_freq_handles_checkpoint_at_endrrrrr rr ScalingConfig RunConfigstrr DataConfigrrintrVrR classmethodr2Boosterr, __classcell__rrrZrr Zs~h      4   ;r )%logging functoolsrtypingrrrrrr2r ray.trainrray.train.constantsr "ray.train.lightgbm._lightgbm_utilsr ray.train.lightgbm.configr ray.train.lightgbm.v2r SimpleLightGBMTrainerray.train.trainerr ray.train.utilsrray.util.annotationsr getLoggerrpr.rTdictrxrzrkrFrrrrs:            <