使用 Ray 分布式 XGBoost

Ray 是一个通用的分布式执行框架。Ray 可用于将计算从单个节点扩展到数百个节点的集群，而无需更改任何代码。

Ray 的 Python 绑定附带了一系列维护良好的机器学习库，用于超参数优化和模型服务。

XGBoost-Ray 项目提供了一个接口，用于在 Ray 集群上运行 XGBoost 训练和预测作业。它允许利用分布式数据表示，例如 Modin 数据帧，以及从云存储（例如 Parquet 文件）进行分布式加载。

XGBoost-Ray 与超参数优化库 Ray Tune 很好地集成，并实现了高级容错处理机制。使用 Ray，只需向集群添加新节点，即可将训练作业扩展到数百个节点。您还可以使用 Ray 来利用多 GPU XGBoost 训练。

安装和启动 Ray

Ray 可以像这样从 PyPI 安装

pip install ray

如果您在单机上使用 Ray，则无需执行任何其他操作 - XGBoost-Ray 在使用时会自动启动本地 Ray 集群。

如果您想在集群上使用 Ray，可以使用 Ray 集群启动器。

安装 XGBoost-Ray

XGBoost-Ray 也通过 PyPI 提供

pip install xgboost_ray

这将安装在 Ray 上运行 XGBoost 所需的所有依赖项，包括 Ray 本身（如果之前尚未安装）。

使用 XGBoost-Ray 进行训练和预测

XGBoost-Ray 使用与核心 XGBoost 相同的 API。只有两个区别

  1. 您将使用 xgboost_ray.RayDMatrix 对象，而不是使用 xgboost.DMatrix。

  2. 有一个额外的 xgboost_ray.RayParams 参数，您可以使用它来配置分布式训练。

简单的训练示例

要运行此简单示例，您需要安装 scikit-learn（使用 pip install sklearn）。

在此示例中，我们将加载 乳腺癌数据集，并使用两个参与者训练一个二元分类器。

from xgboost_ray import RayDMatrix, RayParams, train
from sklearn.datasets import load_breast_cancer

train_x, train_y = load_breast_cancer(return_X_y=True)
train_set = RayDMatrix(train_x, train_y)

evals_result = {}
bst = train(
    {
        "objective": "binary:logistic",
        "eval_metric": ["logloss", "error"],
    },
    train_set,
    evals_result=evals_result,
    evals=[(train_set, "train")],
    verbose_eval=False,
    ray_params=RayParams(num_actors=2, cpus_per_actor=1))

bst.save_model("model.xgb")
print("Final training error: {:.4f}".format(
    evals_result["train"]["error"][-1]))

与非分布式 API 相比，唯一的区别是导入语句（xgboost_ray 而不是 xgboost）、使用 RayDMatrix 而不是 DMatrix，以及传递 xgboost_ray.RayParams 对象。

返回对象是一个常规的 xgboost.Booster 实例。

简单的预测示例

from xgboost_ray import RayDMatrix, RayParams, predict
from sklearn.datasets import load_breast_cancer
import xgboost as xgb

data, labels = load_breast_cancer(return_X_y=True)

dpred = RayDMatrix(data, labels)

bst = xgb.Booster(model_file="model.xgb")
pred_ray = predict(bst, dpred, ray_params=RayParams(num_actors=2))

print(pred_ray)

在此示例中，数据将跨两个参与者进行拆分。结果数组将以正确的顺序集成此数据。

RayParams 对象

RayParams 对象用于配置与分布式训练相关的各种设置。

多 GPU 训练

Ray 自动检测集群节点上的 GPU。为了开始在多个 GPU 上进行训练，您只需设置 RayParams 对象的 gpus_per_actor 参数，以及多 GPU 的 num_actors 参数

ray_params = RayParams(
    num_actors=4,
    gpus_per_actor=1,
)

这将并行训练四个 GPU。

请注意，每个参与者分配一个以上的 GPU 通常没有意义，因为 XGBoost 依赖于 Dask 或 Ray 等分布式库来利用多 GPU 训练。

设置每个参与者的 CPU 数量

XGBoost 本身利用多线程来加速计算。因此，如果您只在 CPU 上进行训练，则每个节点使用一个以上的参与者可能没有好处。在这种情况下，假设您有一个同构节点集群，请将每个参与者的 CPU 数量设置为每个节点上可用的 CPU 数量，并将参与者数量设置为节点数量。

如果您在单个节点上使用多 GPU 训练，请将可用 CPU 数量平均分配给所有参与者。例如，如果您有 16 个 CPU 和 4 个 GPU 可用，则每个参与者应访问 1 个 GPU 和 4 个 CPU。

如果您使用的是异构节点（CPU 数量不同）集群，您可能只想使用每个参与者 CPU 数量的最大公约数。例如，如果您有一个由三个节点组成的集群，分别有 4、8 和 12 个 CPU，那么您将启动 6 个参与者，每个参与者 4 个 CPU，以实现最大的 CPU 利用率。

容错

XGBoost-Ray 支持两种容错模式。在非弹性训练中，每当训练参与者死亡（例如因为节点宕机），训练作业将停止，XGBoost-Ray 将等待参与者（或其资源）再次可用（这可能在不同的节点上），然后所有参与者恢复后继续训练。

在弹性训练中，每当训练参与者死亡时，其余的参与者会继续训练，而无需等待死亡的参与者。如果该参与者恢复，它将重新集成到训练中。

请注意，在弹性训练中，这意味着您将在一段时间内使用较少的数据进行训练。好处是，即使一个节点在训练运行的剩余时间内消失，您也可以继续训练，而不必等待它再次启动。实际上，这通常会导致准确性略有下降，但与非弹性训练相比，训练时间大大缩短。

这两种训练模式都可以使用相应的 xgboost_ray.RayParams 参数进行配置。

超参数优化

XGBoost-Ray 与超参数优化框架 Ray Tune 很好地集成。Ray Tune 使用 Ray 启动多个具有不同超参数配置的分布式试验。如果与 XGBoost-Ray 一起使用，这些试验将启动自己的分布式训练作业。

XGBoost-Ray 自动将评估结果报告回 Ray Tune。您只需要做几件事

  1. 将您的 XGBoost-Ray 训练调用放入一个接受参数配置的函数中（以下示例中的 train_model）。

  2. 创建 xgboost_ray.RayParams 对象（以下示例中的 ray_params）。

  3. 定义参数搜索空间（以下示例中的 config 字典）。

  4. 调用 tune.run()

    • metric 参数应包含您希望优化的指标。通常，这由传递给 xgboost_ray.train() 的 evals 参数的前缀和 XGBoost 参数中传递的 eval_metric 组成（以下示例中的 train-error）。

    • mode 应为 min 或 max，具体取决于您希望最小化还是最大化指标

    • 应使用 ray_params.get_tune_resources() 设置 resources_per_actor。这将确保每个试验都具有启动其分布式训练作业所需的资源。

from xgboost_ray import RayDMatrix, RayParams, train
from sklearn.datasets import load_breast_cancer

num_actors = 4
num_cpus_per_actor = 1

ray_params = RayParams(
    num_actors=num_actors, cpus_per_actor=num_cpus_per_actor)

def train_model(config):
    train_x, train_y = load_breast_cancer(return_X_y=True)
    train_set = RayDMatrix(train_x, train_y)

    evals_result = {}
    bst = train(
        params=config,
        dtrain=train_set,
        evals_result=evals_result,
        evals=[(train_set, "train")],
        verbose_eval=False,
        ray_params=ray_params)
    bst.save_model("model.xgb")

from ray import tune

# Specify the hyperparameter search space.
config = {
    "tree_method": "approx",
    "objective": "binary:logistic",
    "eval_metric": ["logloss", "error"],
    "eta": tune.loguniform(1e-4, 1e-1),
    "subsample": tune.uniform(0.5, 1.0),
    "max_depth": tune.randint(1, 9)
}

# Make sure to use the `get_tune_resources` method to set the `resources_per_trial`
analysis = tune.run(
    train_model,
    config=config,
    metric="train-error",
    mode="min",
    num_samples=4,
    resources_per_trial=ray_params.get_tune_resources())
print("Best hyperparameters", analysis.best_config)

Ray Tune 支持各种搜索算法和库（例如 BayesOpt、Tree-Parzen 估计器）、智能调度器（如逐次减半）以及其他功能。有关更多信息，请参阅Ray Tune 文档。

附加资源