使用PySpark进行大数据处理与机器学习实战指南

1. 技术介绍

1.1 PySpark概述

PySpark是Apache Spark的Python API，它结合了Python的易用性和Spark的分布式计算能力，能够高效处理PB级数据集。Spark基于内存计算的特性使其比传统Hadoop MapReduce快10-100倍，支持流处理、SQL查询、机器学习和图计算。

核心组件：

SparkContext: 应用程序的入口点
RDD（弹性分布式数据集）: 不可变的分布式对象集合
DataFrame: 结构化数据集，支持SQL查询
MLlib: 可扩展的机器学习库
Spark SQL: 结构化数据处理模块

Spark架构图

1.2 技术优势

分布式内存计算引擎
支持批处理和流处理
丰富的生态系统（SQL、ML、GraphX）
容错机制（Lineage记录）
与Hadoop生态无缝集成

2. 实战案例：数据清洗与机器学习

2.1 环境配置

# 安装PySpark
!pip install pyspark# 初始化SparkSession
from pyspark.sql import SparkSession
spark = SparkSession.builder \.appName("PySparkDemo") \.config("spark.driver.memory", "4g") \.getOrCreate()

2.2 数据预处理

# 读取CSV数据
from pyspark.sql.functions import coldf = spark.read.csv("iris.csv", header=True, inferSchema=True)# 数据清洗示例
cleaned_df = df.filter((col("sepal_length") > 0) &(col("sepal_width") < 10)
)# 特征工程
from pyspark.ml.feature import VectorAssemblerassembler = VectorAssembler(inputCols=["sepal_length", "sepal_width", "petal_length", "petal_width"],outputCol="features"
)processed_df = assembler.transform(cleaned_df)# 查看数据模式
processed_df.printSchema()

2.3 机器学习建模

from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml import Pipeline# 划分训练测试集
train_df, test_df = processed_df.randomSplit([0.8, 0.2], seed=42)# 构建Pipeline
lr = LogisticRegression(featuresCol="features", labelCol="species")
pipeline = Pipeline(stages=[lr])# 训练模型
model = pipeline.fit(train_df)# 预测评估
predictions = model.transform(test_df)
evaluator = MulticlassClassificationEvaluator(labelCol="species", predictionCol="prediction",metricName="accuracy"
)accuracy = evaluator.evaluate(predictions)
print(f"Test Accuracy = {accuracy:.4f}")

3. 运行结果

3.1 数据展示

+------------+-----------+------------+-----------+-------+
|sepal_length|sepal_width|petal_length|petal_width|species|
+------------+-----------+------------+-----------+-------+
|         5.1|        3.5|         1.4|        0.2| setosa|
|         4.9|        3.0|         1.4|        0.2| setosa|
|         4.7|        3.2|         1.3|        0.2| setosa|
+------------+-----------+------------+-----------+-------+

3.2 聚合统计

df.groupBy("species").agg({"sepal_length": "avg", "petal_length": "max"}
).show()

输出结果：

+-------+------------------+------------------+
|species| avg(sepal_length)| max(petal_length)|
+-------+------------------+------------------+
| setosa|             5.006|               1.9|
|versicolor|             5.936|              4.9|
|virginica|             6.588|              6.9|
+-------+------------------+------------------+