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: 结构化数据处理模块

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|
+-------+------------------+------------------+

3.3 模型评估

Test Accuracy = 0.967

---

4. 总结与展望

4.1 技术优势总结

• 开发效率：Python语法简洁，API设计直观

• 处理能力：轻松应对TB级数据处理

• 统一平台：SQL查询、流处理、机器学习一站式解决

• 扩展性：支持YARN/Kubernetes等多种集群管理器

4.2 典型应用场景

1. 实时日志分析

2. 用户行为预测

3. 大规模ETL处理

4. 推荐系统构建

5. 金融风控建模

4.3 优化建议

• 合理设置分区数（通常为CPU核心数的2-3倍）

• 使用缓存策略df.cache()复用中间结果

• 避免使用UDF（用户自定义函数）

• 选择合适序列化方式（Kryo Serialization）

4.4 学习路线

1. 掌握RDD基本操作

2. 学习DataFrame API

3. 理解Spark SQL优化原理

4. 实践Structured Streaming

5. 探索GraphFrames图计算

随着Spark 3.0版本的发布，新增的Adaptive Query Execution（AQE）和Dynamic Partition Pruning（DPP）等特性进一步提升了性能。建议持续关注官方文档更新，掌握最新的优化技术。