NVIDIA cuML与RAPIDS GPU机器学习工作流实战教程
Implementation of Machine Learning Workflows with NVIDIA cuML, RAPIDS, GPU Benchmarking, Explainability, Clustering, and Model Inference
推荐理由
给做AI工程落地的同学一份完整的RAPIDS cuML实操手册，含从零配置环境、零代码迁移sklearn、以及详细的GPU/CPU性能基准对比，直接照着跑就能优化链路延迟。
In this tutorial, we implement NVIDIA cuML as a GPU-accelerated machine learning framework and build a practical workflow that demonstrates how RAPIDS can accelerate familiar data science and machine learning tasks. We begin by configuring the GPU environment and examining cuml.accel, which lets us accelerate existing scikit-learn workloads with minimal code changes, before moving to the native cuML API for direct CuPy and cuDF interoperability. We then benchmark CPU and GPU implementations of PCA, K-Means, nearest-neighbor search, logistic regression, random forests, and DBSCAN, while using synchronized timing to obtain meaningful performance measurements. We also build GPU-based manifold-learning and clustering pipelines with UMAP, t-SNE, HDBSCAN, and trustworthiness metrics; explore high-throughput forest inference with FIL; validate GPU-generated SHAP explanations; perform hyperparameter optimization with scikit-learn meta-estimators; and finally serialize trained models while examining portability between GPU and CPU environments.
在本教程中，我们将 NVIDIA cuML 实现为 GPU 加速的机器学习框架，并构建一个实用工作流，展示 RAPIDS 如何加速熟悉的数据科学和机器学习任务。我们首先配置 GPU 环境并检查 cuml.accel，它允许我们通过最少的代码更改来加速现有的 scikit-learn 工作负载，然后转向原生 cuML API 以实现与 CuPy 和 cuDF 的直接互操作性。接着，我们对 PCA、K-Means、最近邻搜索、逻辑回归、随机森林和 DBSCAN 的 CPU 和 GPU 实现进行基准测试，同时使用同步计时以获得有意义的性能测量结果。我们还构建了基于 GPU 的流形学习和聚类管道，使用 UMAP、t-SNE、HDBSCAN 和可信度指标；探索 FIL 的高吞吐量森林推理；验证 GPU 生成的 SHAP 解释；使用 scikit-learn 元估计器执行超参数优化；最后序列化训练好的模型，并考察 GPU 和 CPU 环境之间的可移植性。
代码 · 159 行
import os
import sys
import time
import json
import shutil
import warnings
import subprocess
import importlib
import traceback
warnings.filterwarnings("ignore")
QUICK = False
SEED = 42
SCALE = 0.25 if QUICK else 1.0
N_MAIN = int(200_000 * SCALE)
D_MAIN = 64
N_RF = int(50_000 * SCALE)
D_RF = 32
N_NN_INDEX = int(50_000 * SCALE)
N_NN_QUERY = int(5_000 * SCALE)
N_DBSCAN = int(20_000 * SCALE)
N_MANIFOLD = int(60_000 * SCALE)
N_ACCEL = int(80_000 * SCALE)
RESULTS = []
NOTES = []
def banner(title):
line = "=" * 78
print(f"\n{line}\n {title}\n{line}", flush=True)
def section(title, fn, *args, **kwargs):
banner(title)
t0 = time.perf_counter()
try:
fn(*args, **kwargs)
except Exception:
print(f"[!] Section skipped due to an error:\n{traceback.format_exc()}")
print(f"[section wall time: {time.perf_counter() - t0:.1f}s]", flush=True)
def bootstrap():
if shutil.which("nvidia-smi") is None:
raise SystemExit(
"No NVIDIA GPU found. In Colab: Runtime > Change runtime type > GPU."
)
print(subprocess.run(
["nvidia-smi",
"--query-gpu=name,memory.total,compute_cap,driver_version",
"--format=csv"],
capture_output=True, text=True).stdout)
try:
import cuml
print("cuML already available — skipping install.")
except ImportError:
print("Installing RAPIDS cuML (this takes ~1-3 minutes)...")
pin = ""
try:
import cudf
major_minor = ".".join(cudf.__version__.split("+")[0].split(".")[:2])
pin = f"=={major_minor}.*"
print(f" Pinning to the preinstalled cuDF line: cuml-cu12{pin}")
except Exception:
print(" cuDF not found; installing the latest stable cuml-cu12.")
cmd = [sys.executable, "-m", "pip", "install", "-q",
"--extra-index-url=https://pypi.nvidia.com", f"cuml-cu12{pin}"]
print("$ " + " ".join(cmd))
rc = subprocess.run(cmd).returncode
if rc != 0:
raise SystemExit(
"pip install failed. Alternative that always works on Colab:\n"
" !git clone https://github.com/rapidsai/rapidsai-csp-utils.git\n"
" !python rapidsai-csp-utils/colab/pip-install.py"
)
importlib.invalidate_caches()
import cuml
import cupy
print(f"cuml {cuml.__version__}")
print(f"cupy {cupy.__version__}")
try:
import cudf
print(f"cudf {cudf.__version__}")
except Exception:
pass
import sklearn
print(f"sklearn {sklearn.__version__} (cuML requires scikit-learn >= 1.6)")
bootstrap()
import numpy as np
import cupy as cp
import cuml
import matplotlib.pyplot as plt
from cuml.datasets import make_classification as gpu_make_classification
from cuml.datasets import make_blobs as gpu_make_blobs
rng = np.random.RandomState(SEED)
cp.random.seed(SEED)
class Timer:
def __init__(self, label, sync=True):
self.label = label
self.sync = sync
def __enter__(self):
if self.sync:
cp.cuda.runtime.deviceSynchronize()
self.t0 = time.perf_counter()
return self
def __exit__(self, *exc):
if self.sync:
cp.cuda.runtime.deviceSynchronize()
self.dt = time.perf_counter() - self.t0
print(f" {self.label:<44s} {self.dt:8.3f}s")
return False
def to_numpy(a):
if isinstance(a, cp.ndarray):
return cp.asnumpy(a)
if hasattr(a, "to_numpy"):
return a.to_numpy()
return np.asarray(a)
def record(task, cpu_s, gpu_s):
RESULTS.append((task, cpu_s, gpu_s))
if cpu_s and gpu_s:
print(f" -> {task}: {cpu_s / gpu_s:.1f}x speedup\n")
ACCEL_SCRIPT = f'''
import time
import numpy as np
from sklearn.datasets import make_blobs
from sklearn.decomposition import PCA
from sklearn.cluster import KMeans
from sklearn.neighbors import NearestNeighbors
from sklearn.linear_model import Ridge
X, y = make_blobs(n_samples={N_ACCEL}, n_features=32, centers=12, random_state=0)
X = X.astype("float32"); y = y.astype("float32")
t0 = time.perf_counter()
PCA(n_components=8).fit_transform(X)
KMeans(n_clusters=12, n_init=1, random_state=0).fit(X)
NearestNeighbors(n_neighbors=8).fit(X[:{N_ACCEL // 2}]).kneighbors(X[:5000])
Ridge(alpha=1.0).fit(X, y)
Ridge(alpha=1.0, positive=True).fit(X[:5000], y[:5000])
print("MODELTIME %.3f" % (time.perf_counter() - t0))
'''
def demo_accel():
path = "/content/_accel_demo.py" if os.path.isdir("/content") else "_accel_demo.py"
with open(path, "w") as f:
f.write(ACCEL_SCRIPT)
def run(cmd, label):
print(f"\n$ {' '.join(cmd[1:])}")
t0 = time.perf_counter()
p = subprocess.run(cmd, capture_output=True, text=True)
wall = time.perf_counter() - t0
out = p.stdout + p.stderr
model_s = None
for line in out.splitlines():
if line.startswith("MODELTIME"):
model_s = float(line.split()[1])
print(out.strip()[:4000])
print(f"[{label}] model time = {model_s}s | process wall = {wall:.1f}s")
return model_s
cpu_s = run([sys.executable, path], "stock sklearn")
cmd = [sys.executable, "-m", "cuml.accel", "--profile", path]
gpu_s = run(cmd, "cuml.accel")
if gpu_s is None:
gpu_s = run([sys.executable, "-m", "cuml.accel", path], "cuml.accel")
record("cuml.accel (sklearn script, unmodified)", cpu_s, gpu_s)
NOTES.append(
"cuml.accel needed ZERO source changes; the profile table above shows "
"which calls ran on GPU and why Ridge(positive=True) fell back to CPU."
)代码 · 159 行
import os
import sys
import time
import json
import shutil
import warnings
import subprocess
import importlib
import traceback
warnings.filterwarnings("ignore")
QUICK = False
SEED = 42
SCALE = 0.25 if QUICK else 1.0
N_MAIN = int(200_000 * SCALE)
D_MAIN = 64
N_RF = int(50_000 * SCALE)
D_RF = 32
N_NN_INDEX = int(50_000 * SCALE)
N_NN_QUERY = int(5_000 * SCALE)
N_DBSCAN = int(20_000 * SCALE)
N_MANIFOLD = int(60_000 * SCALE)
N_ACCEL = int(80_000 * SCALE)
RESULTS = []
NOTES = []
def banner(title):
line = "=" * 78
print(f"\n{line}\n {title}\n{line}", flush=True)
def section(title, fn, *args, **kwargs):
banner(title)
t0 = time.perf_counter()
try:
fn(*args, **kwargs)
except Exception:
print(f"[!] Section skipped due to an error:\n{traceback.format_exc()}")
print(f"[section wall time: {time.perf_counter() - t0:.1f}s]", flush=True)
def bootstrap():
if shutil.which("nvidia-smi") is None:
raise SystemExit(
"No NVIDIA GPU found. In Colab: Runtime > Change runtime type > GPU."
)
print(subprocess.run(
["nvidia-smi",
"--query-gpu=name,memory.total,compute_cap,driver_version",
"--format=csv"],
capture_output=True, text=True).stdout)
try:
import cuml
print("cuML already available — skipping install.")
except ImportError:
print("Installing RAPIDS cuML (this takes ~1-3 minutes)...")
pin = ""
try:
import cudf
major_minor = ".".join(cudf.__version__.split("+")[0].split(".")[:2])
pin = f"=={major_minor}.*"
print(f" Pinning to the preinstalled cuDF line: cuml-cu12{pin}")
except Exception:
print(" cuDF not found; installing the latest stable cuml-cu12.")
cmd = [sys.executable, "-m", "pip", "install", "-q",
"--extra-index-url=https://pypi.nvidia.com", f"cuml-cu12{pin}"]
print("$ " + " ".join(cmd))
rc = subprocess.run(cmd).returncode
if rc != 0:
raise SystemExit(
"pip install failed. Alternative that always works on Colab:\n"
" !git clone https://github.com/rapidsai/rapidsai-csp-utils.git\n"
" !python rapidsai-csp-utils/colab/pip-install.py"
)
importlib.invalidate_caches()
import cuml
import cupy
print(f"cuml {cuml.__version__}")
print(f"cupy {cupy.__version__}")
try:
import cudf
print(f"cudf {cudf.__version__}")
except Exception:
pass
import sklearn
print(f"sklearn {sklearn.__version__} (cuML requires scikit-learn >= 1.6)")
bootstrap()
import numpy as np
import cupy as cp
import cuml
import matplotlib.pyplot as plt
from cuml.datasets import make_classification as gpu_make_classification
from cuml.datasets import make_blobs as gpu_make_blobs
rng = np.random.RandomState(SEED)
cp.random.seed(SEED)
class Timer:
def __init__(self, label, sync=True):
self.label = label
self.sync = sync
def __enter__(self):
if self.sync:
cp.cuda.runtime.deviceSynchronize()
self.t0 = time.perf_counter()
return self
def __exit__(self, *exc):
if self.sync:
cp.cuda.runtime.deviceSynchronize()
self.dt = time.perf_counter() - self.t0
print(f" {self.label:<44s} {self.dt:8.3f}s")
return False
def to_numpy(a):
if isinstance(a, cp.ndarray):
return cp.asnumpy(a)
if hasattr(a, "to_numpy"):
return a.to_numpy()
return np.asarray(a)
def record(task, cpu_s, gpu_s):
RESULTS.append((task, cpu_s, gpu_s))
if cpu_s and gpu_s:
print(f" -> {task}: {cpu_s / gpu_s:.1f}x speedup\n")
ACCEL_SCRIPT = f'''
import time
import numpy as np
from sklearn.datasets import make_blobs
from sklearn.decomposition import PCA
from sklearn.cluster import KMeans
from sklearn.neighbors import NearestNeighbors
from sklearn.linear_model import Ridge
X, y = make_blobs(n_samples={N_ACCEL}, n_features=32, centers=12, random_state=0)
X = X.astype("float32"); y = y.astype("float32")
t0 = time.perf_counter()
PCA(n_components=8).fit_transform(X)
KMeans(n_clusters=12, n_init=1, random_state=0).fit(X)
NearestNeighbors(n_neighbors=8).fit(X[:{N_ACCEL // 2}]).kneighbors(X[:5000])
Ridge(alpha=1.0).fit(X, y)
Ridge(alpha=1.0, positive=True).fit(X[:5000], y[:5000])
print("MODELTIME %.3f" % (time.perf_counter() - t0))
'''
def demo_accel():
path = "/content/_accel_demo.py" if os.path.isdir("/content") else "_accel_demo.py"
with open(path, "w") as f:
f.write(ACCEL_SCRIPT)
def run(cmd, label):
print(f"\n$ {' '.join(cmd[1:])}")
t0 = time.perf_counter()
p = subprocess.run(cmd, capture_output=True, text=True)
wall = time.perf_counter() - t0
out = p.stdout + p.stderr
model_s = None
for line in out.splitlines():
if line.startswith("MODELTIME"):
model_s = float(line.split()[1])
print(out.strip()[:4000])
print(f"[{label}] model time = {model_s}s | process wall = {wall:.1f}s")
return model_s
cpu_s = run([sys.executable, path], "stock sklearn")
cmd = [sys.executable, "-m", "cuml.accel", "--profile", path]
gpu_s = run(cmd, "cuml.accel")
if gpu_s is None:
gpu_s = run([sys.executable, "-m", "cuml.accel", path], "cuml.accel")
record("cuml.accel (sklearn script, unmodified)", cpu_s, gpu_s)
NOTES.append(
"cuml.accel needed ZERO source changes; the profile table above shows "
"which calls ran on GPU and why Ridge(positive=True) fell back to CPU."
)We configure the tutorial environment, define dataset sizes and benchmarking utilities, and verify that an NVIDIA GPU is available. We install and initialize RAPIDS cuML when necessary, set up CuPy and reproducibility controls, and create synchronized timing and result-tracking helpers. We also demonstrate cuml.accel by running an unmodified scikit-learn workload and comparing its CPU execution with GPU-accelerated execution.
我们配置教程环境，定义数据集大小和基准测试工具，并验证是否有可用的 NVIDIA GPU。在必要时安装并初始化 RAPIDS cuML，设置 CuPy 和可重复性控制，并创建同步计时和结果跟踪辅助函数。我们还通过运行未修改的 scikit-learn 工作负载并比较其 CPU 执行与 GPU 加速执行来演示 cuml.accel。
更进一步：量化金融体系
看懂新闻只是起点——沿量化金融路径，把它变成能交付的工程能力