使用 dask-ml 和 cuml 进行 HPO#
引言#
超参数优化是选择模型超参数值的任务，这些超参数值能够在特定测试数据集上提供问题的最优结果。这通常是关键的一步，如果操作得当，可以帮助提高模型精度。交叉验证常用于更准确地估计搜索过程中模型的性能。交叉验证是将训练集分割成互补的子集，在一个子集上进行训练，然后在另一个子集上预测模型性能的方法。这潜在地表明了模型将如何泛化到它以前未见过的数据上。
尽管超参数优化具有理论重要性，但由于运行如此多不同训练任务所需的资源，在实际应用中很难实现。
我们将在此 Notebook 中探讨的两种方法是
1. 网格搜索 (GridSearch)#
顾名思义，“搜索”是在用户提供的参数网格中对每种可能的组合进行的。用户必须手动定义此网格。对于需要调优的每个参数，都提供了一组值，最终的网格搜索是对从每个集合中选取一个元素组成的元组进行，因此产生了元素的笛卡尔积。
例如，假设我们要对 XGBoost 执行 HPO。为简单起见，我们仅调优 n_estimators 和 max_depth
n_estimators: [50, 100, 150]max_depth: [6, 7, ,8]
网格搜索将在 |n_estimators| x |max_depth| 上进行，即 3 x 3 = 9。正如您可能猜到的，随着参数数量及其搜索空间的增加，网格大小会迅速增长。
2. 随机搜索 (RandomSearch)#
随机搜索用在指定空间内随机选择参数来取代之前搜索的穷举性质。在影响模型性能的参数数量较少（低维优化问题）的情况下，这种方法可以胜过网格搜索。由于它不从笛卡尔积中选取所有元组，因此往往能更快地得出结果，并且性能可以与网格搜索方法相媲美。值得记住的是，这种搜索的随机性意味着每次运行的结果可能会有所不同。
其他一些用于 HPO 的方法包括
贝叶斯优化
基于梯度的优化
进化优化
要了解更多关于 HPO 的信息，Notebook 末尾链接了一些论文供进一步阅读。
现在我们对 HPO 有了基本的了解，让我们讨论一下我们希望通过这个演示实现什么。本 Notebook 的目标是展示超参数优化的重要性以及 dask-ml GPU 在 xgboost 和 cuML-RF 上的性能。
在此演示中，我们将使用 Airline dataset。问题的目标是预测到达延迟。它包含约 1.16 亿条记录，有 13 个属性用于确定给定航空公司的延迟。我们将此问题修改为二元分类问题，以确定航班是否会延迟 (True) 或不会延迟 (False)。
开始吧！
import warnings
warnings.filterwarnings("ignore") # Reduce number of messages/warnings displayed
import os
from urllib.request import urlretrieve
import cudf
import dask_ml.model_selection as dcv
import numpy as np
import pandas as pd
import xgboost as xgb
from cuml.ensemble import RandomForestClassifier
from cuml.metrics.accuracy import accuracy_score
from cuml.model_selection import train_test_split
from dask.distributed import Client
from dask_cuda import LocalCUDACluster
from sklearn.metrics import make_scorer
启动 CUDA 集群#
本 Notebook 设计用于在具有多个 GPU 的单节点上运行，您可以从 AWS、GCP、Azure、IBM 等获取多 GPU 虚拟机。
我们启动一个本地集群，并使其准备好使用 dask 运行分布式任务。
下面，LocalCUDACluster 为当前系统中的每个 GPU 启动一个 Dask worker。它是 RAPIDS 项目的一部分。了解更多
cluster = LocalCUDACluster()
client = Client(cluster)
client
数据准备#
我们下载 Airline dataset 并将其保存到由 data_dir 和 file_name 指定的本地目录。在此步骤中，我们还将输入数据转换为适当的 dtypes。为此，我们将使用 prepare_dataset 函数。
注意：为确保本示例在配置适中的机器上快速运行，我们默认使用 Airline dataset 的一个小子集。要使用完整数据集，请将参数 use_full_dataset=True 传递给 prepare_dataset 函数。
data_dir = "./rapids_hpo/data/"
file_name = "airlines.parquet"
parquet_name = os.path.join(data_dir, file_name)
parquet_name
def prepare_dataset(use_full_dataset=False):
global file_path, data_dir
if use_full_dataset:
url = "https://data.rapids.ai/cloud-ml/airline_20000000.parquet"
else:
url = "https://data.rapids.ai/cloud-ml/airline_small.parquet"
if os.path.isfile(parquet_name):
print(f" > File already exists. Ready to load at {parquet_name}")
else:
# Ensure folder exists
os.makedirs(data_dir, exist_ok=True)
def data_progress_hook(block_number, read_size, total_filesize):
if (block_number % 1000) == 0:
print(
f" > percent complete: { 100 * ( block_number * read_size ) / total_filesize:.2f}\r",
end="",
)
return
urlretrieve(
url=url,
filename=parquet_name,
reporthook=data_progress_hook,
)
print(f" > Download complete {file_name}")
input_cols = [
"Year",
"Month",
"DayofMonth",
"DayofWeek",
"CRSDepTime",
"CRSArrTime",
"UniqueCarrier",
"FlightNum",
"ActualElapsedTime",
"Origin",
"Dest",
"Distance",
"Diverted",
]
dataset = cudf.read_parquet(parquet_name)
# encode categoricals as numeric
for col in dataset.select_dtypes(["object"]).columns:
dataset[col] = dataset[col].astype("category").cat.codes.astype(np.int32)
# cast all columns to int32
for col in dataset.columns:
dataset[col] = dataset[col].astype(np.float32) # needed for random forest
# put target/label column first [ classic XGBoost standard ]
output_cols = ["ArrDelayBinary"] + input_cols
dataset = dataset.reindex(columns=output_cols)
return dataset
df = prepare_dataset()
import time
from contextlib import contextmanager
# Helping time blocks of code
@contextmanager
def timed(txt):
t0 = time.time()
yield
t1 = time.time()
print("%32s time: %8.5f" % (txt, t1 - t0))
# Define some default values to make use of across the notebook for a fair comparison
N_FOLDS = 5
N_ITER = 25
label = "ArrDelayBinary"
分割数据#
我们使用 cuml train_test_split 将数据随机分割为训练集和测试集，并创建数据的 CPU 版本。
X_train, X_test, y_train, y_test = train_test_split(df, label, test_size=0.2)
X_cpu = X_train.to_pandas()
y_cpu = y_train.to_numpy()
X_test_cpu = X_test.to_pandas()
y_test_cpu = y_test.to_numpy()
设置自定义 cuML scorers#
scikit-learn 和 dask-ml 的搜索函数（如 GridSearchCV）期望指标函数（如 accuracy_score）与“scorer”API 匹配。这可以使用 scikit-learn 的 make_scorer 函数来实现。
我们将使用 cuML accuracy_score 函数生成一个 cuml_scorer。您还会注意到一个 accuracy_score_wrapper，它主要将 y 标签转换为 float32 类型。这是因为目前一些 cuML 模型只接受这种类型，为了使其兼容，我们执行了此转换。
我们还创建了用于以 2 种不同模式执行 HPO 的辅助函数
gpu-grid: 执行基于 GPU 的 GridSearchCVgpu-random: 执行基于 GPU 的 RandomizedSearchCV
def accuracy_score_wrapper(y, y_hat):
"""
A wrapper function to convert labels to float32,
and pass it to accuracy_score.
Params:
- y: The y labels that need to be converted
- y_hat: The predictions made by the model
"""
y = y.astype("float32") # cuML RandomForest needs the y labels to be float32
return accuracy_score(y, y_hat, convert_dtype=True)
accuracy_wrapper_scorer = make_scorer(accuracy_score_wrapper)
cuml_accuracy_scorer = make_scorer(accuracy_score, convert_dtype=True)
def do_HPO(model, gridsearch_params, scorer, X, y, mode="gpu-Grid", n_iter=10):
"""
Perform HPO based on the mode specified
mode: default gpu-Grid. The possible options are:
1. gpu-grid: Perform GPU based GridSearchCV
2. gpu-random: Perform GPU based RandomizedSearchCV
n_iter: specified with Random option for number of parameter settings sampled
Returns the best estimator and the results of the search
"""
if mode == "gpu-grid":
print("gpu-grid selected")
clf = dcv.GridSearchCV(model, gridsearch_params, cv=N_FOLDS, scoring=scorer)
elif mode == "gpu-random":
print("gpu-random selected")
clf = dcv.RandomizedSearchCV(
model, gridsearch_params, cv=N_FOLDS, scoring=scorer, n_iter=n_iter
)
else:
print("Unknown Option, please choose one of [gpu-grid, gpu-random]")
return None, None
res = clf.fit(X, y)
print(f"Best clf and score {res.best_estimator_} {res.best_score_}\n---\n")
return res.best_estimator_, res
def print_acc(model, X_train, y_train, X_test, y_test, mode_str="Default"):
"""
Trains a model on the train data provided, and prints the accuracy of the trained model.
mode_str: User specifies what model it is to print the value
"""
y_pred = model.fit(X_train, y_train).predict(X_test)
score = accuracy_score(y_pred, y_test.astype("float32"), convert_dtype=True)
print(f"{mode_str} model accuracy: {score}")
X_train.shape
启动 HPO#
我们将首先查看模型在未进行网格搜索时的性能，然后将其与搜索后的性能进行比较。
XGBoost#
为了执行超参数优化，我们使用了 XGBClassifier 的 sklearn 版本。我们使用这个版本是为了使其兼容并易于与 scikit-learn 版本进行比较。该模型接受可以在文档中找到的一组参数。我们主要关注 max_depth、learning_rate、min_child_weight、reg_alpha 和 num_round，因为这些参数对 XGBoost 的性能影响最大。
在此处阅读更多关于这些参数的用途：这里
默认性能#
我们首先使用具有默认参数的模型，并查看模型的精度。在本例中，精度为 84%。
model_gpu_xgb_ = xgb.XGBClassifier(tree_method="gpu_hist")
print_acc(model_gpu_xgb_, X_train, y_cpu, X_test, y_test_cpu)
参数分布#
我们定义网格以执行搜索的方式是包含需要用于搜索的参数范围。在本例中，我们使用 np.arange，它返回等间距值的 ndarray；np.logspace 返回在对数尺度上等间距的指定数量的样本。我们也可以指定为列表、NumPy 数组或使用任何在调用时提供样本的随机变量样本。SciPy 也为此提供了各种函数。
# For xgb_model
model_gpu_xgb = xgb.XGBClassifier(tree_method="gpu_hist")
# More range
params_xgb = {
"max_depth": np.arange(start=3, stop=12, step=3), # Default = 6
"alpha": np.logspace(-3, -1, 5), # default = 0
"learning_rate": [0.05, 0.1, 0.15], # default = 0.3
"min_child_weight": np.arange(start=2, stop=10, step=3), # default = 1
"n_estimators": [100, 200, 1000],
}
RandomizedSearchCV#
我们现在将尝试 RandomizedSearchCV。n_iter 指定搜索需要执行的参数点数。在这里，我们将搜索 N_ITER（前面已定义）个点以获得最佳性能。
mode = "gpu-random"
with timed("XGB-" + mode):
res, results = do_HPO(
model_gpu_xgb,
params_xgb,
cuml_accuracy_scorer,
X_train,
y_cpu,
mode=mode,
n_iter=N_ITER,
)
num_params = len(results.cv_results_["mean_test_score"])
print(f"Searched over {num_params} parameters")
print_acc(res, X_train, y_cpu, X_test, y_test_cpu, mode_str=mode)
mode = "gpu-grid"
with timed("XGB-" + mode):
res, results = do_HPO(
model_gpu_xgb, params_xgb, cuml_accuracy_scorer, X_train, y_cpu, mode=mode
)
num_params = len(results.cv_results_["mean_test_score"])
print(f"Searched over {num_params} parameters")
print_acc(res, X_train, y_cpu, X_test, y_test_cpu, mode_str=mode)
改进的性能#
性能提高了 5%。
我们注意到，尽管随机搜索只使用了 25 种参数组合，但执行网格搜索和随机搜索产生的性能提升相似。我们将继续使用 RF 执行随机搜索，假设如果范围足够大，性能不会有显著差异。
可视化搜索结果#
让我们绘制一些图表来了解参数如何影响精度。这些图表的代码包含在 cuml/experimental/hyperopt_utils/plotting_utils.py 中。
测试分数的平均值/标准差#
我们为每个图表固定除一个参数外的所有参数，并绘制该参数对平均测试分数的影响，误差条表示标准差。
from cuml.experimental.hyperopt_utils import plotting_utils
plotting_utils.plot_search_results(results)
热力图#
参数对之间（我们可以组合所有可能的对，但此 Notebook 中仅显示一对）
这直观地展示了这对参数如何影响测试分数。
df_gridsearch = pd.DataFrame(results.cv_results_)
plotting_utils.plot_heatmap(df_gridsearch, "param_max_depth", "param_n_estimators")
RandomForest (随机森林)#
让我们使用 RandomForest 分类器执行超参数搜索。我们将使用 cuml RandomForestClassifier 并使用热力图可视化结果。
## Random Forest
model_rf_ = RandomForestClassifier()
params_rf = {
"max_depth": np.arange(start=3, stop=15, step=2), # Default = 6
"max_features": [0.1, 0.50, 0.75, "auto"], # default = 0.3
"n_estimators": [100, 200, 500, 1000],
}
for col in X_train.columns:
X_train[col] = X_train[col].astype("float32")
y_train = y_train.astype("int32")
print(
"Default acc: ",
accuracy_score(model_rf_.fit(X_train, y_train).predict(X_test), y_test),
)
mode = "gpu-random"
model_rf = RandomForestClassifier()
with timed("RF-" + mode):
res, results = do_HPO(
model_rf,
params_rf,
cuml_accuracy_scorer,
X_train,
y_cpu,
mode=mode,
n_iter=N_ITER,
)
num_params = len(results.cv_results_["mean_test_score"])
print(f"Searched over {num_params} parameters")
print("Improved acc: ", accuracy_score(res.predict(X_test), y_test))
df_gridsearch = pd.DataFrame(results.cv_results_)
plotting_utils.plot_heatmap(df_gridsearch, "param_max_depth", "param_n_estimators")
结论与后续步骤#
我们注意到 GridSearch 和 RandomizedSearch 的基本版本性能得到了改进。通常，使用的数据越多，模型性能越好，因此建议您尝试使用更大的数据集和更广泛的参数范围。
该实验也可以使用不同的分类器和不同的参数范围重复进行，以了解 HPO 如何帮助提高性能指标。在本示例中，我们选择了一个基本指标 - 精度，但您可以使用更有趣的指标来帮助确定模型的有用性。您甚至可以将参数列表发送给评分函数。这使得 HPO 异常强大，可以显著提升我们生成的模型。