使用 Dask 和 Scikit-learn 处理大型数据集

📢 转载信息

原文链接：https://www.kdnuggets.com/processing-large-datasets-with-dask-and-scikit-learn

原文作者：Iván Palomares Carrascosa

# 引言

 
Dask 是一个利用并行计算能力的一系列软件包，在处理大型数据集或构建高效、数据密集型应用（如高级分析和机器学习系统）时极其有用。它最显著的优势之一是与现有 Python 框架的无缝集成，包括通过并行化工作流支持与 scikit-learn 模块一起处理大型数据集。本文将揭示如何在硬件资源有限的情况下，利用 Dask 实现可扩展的数据处理。

# 分步演练

 
尽管加州住房数据集（California Housing dataset）并非极其庞大，但其规模适中，非常适合作为温和的、具有说明性的编码示例，用以展示如何联合利用 Dask 和 scikit-learn 来大规模处理数据。

Dask 提供了一个 dataframe 模块，它模仿了 Pandas DataFrame 对象的许多方面，用于处理可能无法完全载入内存的大型数据集。我们将使用这个 Dask DataFrame 结构从 GitHub 存储库中的 CSV 文件加载数据，代码如下：

import dask.dataframe as dd url = "https://raw.githubusercontent.com/gakudo-ai/open-datasets/refs/heads/main/housing.csv" df = dd.read_csv(url) df.head()

这里有一个重要的注意事项。如果你想查看数据集的“形状”——即行数和列数——这个方法比简单地使用 df.shape 要棘手一些。相反，你应该这样做：

num_rows = df.shape[0].compute() num_cols = df.shape[1] print(f"Number of rows: {num_rows}") print(f"Number of columns: {num_cols}")

输出：

Number of rows: 20640 Number of columns: 10

请注意，我们使用了 Dask 的 compute() 来惰性地计算行数，但没有计算列数。数据集的元数据允许我们立即获得列数（特征数），而确定一个（假设）大于内存且已被分区的可能数据集的行数，则需要分布式计算：这是 compute() 为我们透明处理的事情。

数据预处理通常是构建机器学习模型或估计器之前的步骤。在进入建模部分之前，由于本文的重点是展示 Dask 如何用于数据处理，让我们来清洗和准备数据。

数据准备中的一个常见步骤是处理缺失值。使用 Dask，这个过程和使用 Pandas 时一样无缝。例如，下面的代码会删除在任何属性中包含缺失值的实例的行：

df = df.dropna() num_rows = df.shape[0].compute() num_cols = df.shape[1] print(f"Number of rows: {num_rows}") print(f"Number of columns: {num_cols}")

现在数据集已经减少了 200 多个实例，总共有 20433 行。

接下来，我们可以通过引入 scikit-learn 的 StandardScaler 或任何其他合适的缩放方法来缩放数据集中的一些数值特征：

from sklearn.preprocessing import StandardScaler numeric_df = df.select_dtypes(include=["number"]) X_pd = numeric_df.drop("median_house_value", axis=1).compute() scaler = StandardScaler() X_scaled = scaler.fit_transform(X_pd)

值得注意的是，对于我们在一系列数据集密集型操作中（如删除包含缺失值的行，然后删除目标列 "median_house_value"），我们必须在链式操作的末尾添加 compute()。这是因为 Dask 中的数据集转换是惰性执行的。一旦调用了 compute()，对数据集的链式转换结果就会被具体化为一个 Pandas DataFrame（Dask 依赖于 Pandas，因此除非你直接调用仅限 Pandas 的函数，否则你不需要在代码中显式导入 Pandas 库）。

如果我们想训练一个机器学习模型呢？那么我们应该提取目标变量 "median_house_value"，并应用相同的原则将其转换为 Pandas 对象：

y = df["median_house_value"] y_pd = y.compute()

从现在开始，将数据集分割成训练集和测试集、训练像 RandomForestRegressor 这样的回归模型，并在测试数据上评估其误差的过程，完全类似于传统上使用 Pandas 和 scikit-learn 进行的编排式操作。由于基于树的模型对特征缩放不敏感，你可以使用未缩放的特征（X_pd）或缩放后的特征（X_scaled）。下面我们使用上面计算出的缩放后的特征：

from sklearn.model_selection import train_test_split from sklearn.ensemble import RandomForestRegressor from sklearn.metrics import mean_squared_error import numpy as np # Use the scaled feature matrix produced earlier X_train, X_test, y_train, y_test = train_test_split(X_scaled, y_pd, test_size=0.2, random_state=42) model = RandomForestRegressor(n_estimators=100, random_state=42, n_jobs=-1) model.fit(X_train, y_train) y_pred = model.predict(X_test) mse = mean_squared_error(y_test, y_pred) rmse = np.sqrt(mse) print(f"RMSE: {rmse:.2f}")

输出：

RMSE: 49673.99

# 总结

 
Dask 和 scikit-learn 可以协同工作，利用可扩展的、并行化的数据处理工作流，例如高效地预处理大型数据集以构建机器学习模型。本文演示了如何使用 Dask 加载、清洗、准备和转换数据，随后应用标准的 scikit-learn 工具进行机器学习建模——所有这些操作都在处理海量数据集时优化了内存使用并加速了管道（pipeline）。
 
 

Iván Palomares Carrascosa 是人工智能、机器学习、深度学习和 LLM 领域的领导者、作家、演讲者和顾问。他负责培训和指导他人如何在现实世界中利用人工智能。

🚀 想要体验更好更全面的AI调用？

欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。