如何使用 PySpark 进行机器学习-Python教程-PHP中文网

自从 Apache Spark（一种用于处理大数据的开源框架）发布以来，它已成为跨多个容器并行处理大量数据的最广泛使用的技术之一 - 它以效率和速度而自豪与之前存在的类似软件相比。

通过 PySpark 在 Python 中使用这项令人惊叹的技术是可行的，PySpark 是一个 Python API，允许您使用 Python 编程语言与 ApacheSpark 进行交互并挖掘 ApacheSpark 的惊人潜力。

在本文中，您将学习并开始使用 PySpark 使用线性回归算法构建机器学习模型。

注意：预先了解 Python、VSCode 等 IDE、如何使用命令提示符/终端以及熟悉机器学习概念对于正确理解本文中包含的概念至关重要。

通过阅读本文，您应该能够：

了解 ApacheSpark 是什么。
了解 PySpark 以及如何将其用于机器学习。

PySpark 到底是什么？

根据 Apache Spark 官方网站，PySpark 可让您利用 ApacheSpark（简单性、速度、可扩展性、多功能性）和 Python（丰富的生态系统、成熟的库、简单性）的综合优势进行“数据工程”单节点机器或集群上的数据科学和机器学习。”

How to Use PySpark for Machine Learning
图片来源

PySpark 是 ApacheSpark 的 Python API，这意味着它充当一个接口，让用 Python 编写的代码与用 Scala 编写的 ApacheSpark 技术进行通信。这样，已经熟悉Python生态系统的专业人士就可以快速利用ApacheSpark技术。这也确保了 Python 中使用的现有库保持相关性。

有关如何使用 PySpark 进行机器学习的详细指南

在接下来的步骤中，我们将使用线性回归算法构建机器学习模型：

安装项目依赖项：我假设您的计算机上已经安装了 Python。如果没有，请先安装它，然后再进行下一步。打开终端或命令提示符并输入以下代码以安装 PySpark 库。

pip install pyspark

登录后复制

如果没有这些额外的 Python 库，您可以安装它们。

pip install pyspark

登录后复制

创建文件并导入必要的库：打开 VSCode，然后在您选择的项目目录中为您的项目创建一个文件，例如 pyspart_model.py。打开文件并导入项目所需的库。

pip install pandas numpy

登录后复制

创建 Spark 会话：通过在导入下输入此代码来启动项目的 Spark 会话。

from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator
import pandas as pd

登录后复制

读取 CSV 文件（您将使用的数据集）：如果您的项目目录/文件夹中已有名为 data.csv 的数据集，请使用以下代码加载它。

spark = SparkSession.builder.appName("LogisticRegressionExample").getOrCreate()

登录后复制

探索性数据分析：此步骤可帮助您了解正在使用的数据集。检查空值并决定使用的清理方法。

data = spark.read.csv("data.csv", header=True, inferSchema=True)

登录后复制

如果您正在使用小型数据集，您可以将其转换为 Python 数据框和目录，并使用 Python 检查缺失值。

# Display the schema my
 data.printSchema() 
# Show the first ten rows 
data.show(10)
# Count null values in each column
missing_values = df.select(
    [count(when(isnull(c), c)).alias(c) for c in df.columns]
)

# Show the result
missing_values.show()

登录后复制

数据预处理：此步骤涉及将数据集中的列/特征转换为 PySpark 的机器学习库可以轻松理解或兼容的格式。

使用 VectorAssembler 将所有特征组合到单个向量列中。

pandas_df = data.toPandas()
# Use Pandas to check missing values
print(pandas_df.isna().sum())

登录后复制

分割数据集：按照您方便的比例分割数据集。在这里，我们使用 70% 到 30%：70% 用于训练，30% 用于测试模型。

# Combine feature columns into a single vector column
feature_columns = [col for col in data.columns if col != "label"]
assembler = VectorAssembler(inputCols=feature_columns, outputCol="features")

# Transform the data
data = assembler.transform(data)

# Select only the 'features' and 'label' columns for training
final_data = data.select("features", "label")

# Show the transformed data
final_data.show(5)

登录后复制

训练你的模型：我们正在使用逻辑回归算法来训练我们的模型。

创建 LogisticRegression 类的实例并拟合模型。

train_data, test_data = final_data.randomSplit([0.7, 0.3], seed=42)

登录后复制

使用您训练的模型进行预测：使用我们在上一步中训练的模型进行预测

lr = LogisticRegression(featuresCol="features", labelCol="label")

# Train the model
lr_model = lr.fit(train_data)

登录后复制

模型评估：在这里，正在评估模型以确定其预测性能或其正确性水平。我们通过使用合适的评估指标来实现这一目标。

使用 AUC 指标评估模型

predictions = lr_model.transform(test_data)
# Show predictions
predictions.select("features", "label", "prediction", "probability").show(5)

登录后复制

本文使用的端到端代码如下所示：

evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="label", metricName="areaUnderROC")

# Compute the AUC
auc = evaluator.evaluate(predictions)
print(f"Area Under ROC: {auc}")

登录后复制