PySpark，一个超级强大的 Python 库

各人好！我是炒青椒不放辣，关注我，收看每期的编程干货。

一个简单的库，也允许以大概开启我们的智慧之门，
一个平凡的方法，也许能在危急时刻拯救我们于水深火热，
一个新颖的头脑方式，也许能引发我们无尽的创造力，
一个独特的技巧，也许能成为我们的隐形盾牌……

神奇的 Python 库之旅，第 14 章
  
  
一、初识 PySpark

PySpark 是 Apache Spark 的 Python API，它让我们可以大概在 Python 环境中使用 Spark 的强大功能。Spark 是一个快速的、通用的大数据处理引擎，可以大概以分布式的方式处理大规模数据。通过 PySpark，我们可以使用 Spark 的所有功能，包括数据处理、呆板学习、流处理等。

为什么选择 PySpark？

• 高效处理大数据：Spark 的内存盘算能力使得它比传统的 MapReduce 快许多倍。
  
• 丰富的 API：PySpark 提供了丰富的 API，支持各种数据操作和处理。
  
• 与 Hadoop 兼容：PySpark 可以与 Hadoop 生态系统无缝集成，利用 HDFS、Hive 等工具。
  
• 机动性高：PySpark 兼具 Python 的简便和 Spark 的强大功能，得当各种数据处理使命。
  

安装 PySpark
安装 PySpark 非常简单，只需要一行命令：

1. pip install pyspark

复制代码

设置 PySpark
在使用 PySpark 之前，我们需要设置 Spark 环境。确保你已经安装了 Java 和 Spark，并将 Spark 的 bin 目录添加到系统的 PATH 环境变量中。

你可以在 Python 脚本中创建 SparkSession 来启动 Spark 应用：

1. from pyspark.sql import SparkSession
3. # 创建 SparkSession
4. spark = SparkSession.builder \
5.     .appName("PySpark Example") \
6.     .getOrCreate()
8. print("Spark 版本:", spark.version)

复制代码

…
Github 项目地址;
   https://github.com/apache/spark/tree/master/python/pyspark
  …
二、基本操作

创建 RDD
RDD（Resilient Distributed Dataset）是 Spark 的基本数据布局。我们可以通过并行化现有集合或从外部存储读取数据来创建 RDD。

1. # 并行化集合创建 RDD
2. data = [1, 2, 3, 4, 5]
3. rdd = spark.sparkContext.parallelize(data)
5. # 从外部存储读取数据创建 RDD
6. text_rdd = spark.sparkContext.textFile("path/to/file.txt")

复制代码

基本 RDD 操作
RDD 支持多种操作，包括转换操作和举措操作。转换操作返回一个新的 RDD，而举措操作返回一个值。

1. # 转换操作
2. mapped_rdd = rdd.map(lambda x: x * 2)
3. filtered_rdd = rdd.filter(lambda x: x % 2 == 0)
5. # 行动操作
6. collected_data = mapped_rdd.collect()  # 收集所有元素
7. sum_of_elements = rdd.reduce(lambda x, y: x + y)  # 求和
9. print("收集的数据:", collected_data)
10. print("元素和:", sum_of_elements)

复制代码

…
三、DataFrame 和 Spark SQL

DataFrame 是 PySpark 提供的高级数据布局，类似于 Pandas 的 DataFrame，但可以处理大规模数据。Spark SQL 则允许我们使用 SQL 查询来操作 DataFrame。

创建 DataFrame
我们可以通过布局化数据文件（如 CSV、JSON）、现有 RDD 或 Pandas DataFrame 来创建 Spark DataFrame。

1. # 从 CSV 文件创建 DataFrame
2. df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)
4. # 从 JSON 文件创建 DataFrame
5. json_df = spark.read.json("path/to/file.json")
7. # 从 Pandas DataFrame 创建 Spark DataFrame
8. import pandas as pd
10. pandas_df = pd.DataFrame({"name": ["Alice", "Bob"], "age": [25, 30]})
11. spark_df = spark.createDataFrame(pandas_df)

复制代码

基本 DataFrame 操作
DataFrame 支持多种操作，包括选择、过滤、聚合等。

1. 基本 DataFrame 操作
2. DataFrame 支持多种操作，包括选择、过滤、聚合等。

复制代码

使用 Spark SQL
我们可以将 DataFrame 注册为临时视图，并使用 SQL 查询来操作数据。

1. # 注册临时视图
2. df.createOrReplaceTempView("people")
4. # 使用 SQL 查询
5. result = spark.sql("SELECT name, age FROM people WHERE age > 25")
6. result.show()

复制代码

…
四、呆板学习与流处理

PySpark 提供了丰富的呆板学习和流处理功能，分别通过 MLlib 和 Structured Streaming 实现。

呆板学习
MLlib 是 Spark 的呆板学习库，提供了多种算法和工具。让我们看一个简单的呆板学习例子，使用线性回归模型。

1. from pyspark.ml.regression import LinearRegression
3. # 创建训练数据
4. training = spark.createDataFrame([
5.     (1.0, 2.0),
6.     (2.0, 3.0),
7.     (3.0, 4.0),
8.     (4.0, 5.0),
9. ], ["label", "features"])
11. # 创建线性回归模型
12. lr = LinearRegression()
14. # 训练模型
15. model = lr.fit(training)
17. # 打印模型系数和截距
18. print("系数:", model.coefficients)
19. print("截距:", model.intercept)

复制代码

流处理
Structured Streaming 是 Spark 提供的流处理引擎，支持对实时数据流的处理。

1. from pyspark.sql.functions import split
3. # 创建流式 DataFrame
4. lines = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()
6. # 分割行中的单词
7. words = lines.select(split(lines.value, " ").alias("word"))
9. # 计数每个单词出现的次数
10. word_counts = words.groupBy("word").count()
12. # 启动查询并将结果输出到控制台
13. query = word_counts.writeStream.outputMode("complete").format("console").start()
15. query.awaitTermination()

复制代码

…
五、实战案例

处理大规模日记数据
让我们用 PySpark 处理大规模日记数据，盘算每个 IP 地址的访问次数。

1. # 读取日志文件
2. log_df = spark.read.text("path/to/log/file")
4. # 提取 IP 地址
5. ip_df = log_df.select(split(log_df.value, " ")[0].alias("ip"))
7. # 计算每个 IP 地址的访问次数
8. ip_counts = ip_df.groupBy("ip").count().orderBy("count", ascending=False)
10. # 显示结果
11. ip_counts.show()

复制代码

呆板学习分类
让我们用 PySpark 进行一个简单的呆板学习分类使命，使用决策树分类器。

1. from pyspark.ml.classification import DecisionTreeClassifier
2. from pyspark.ml.feature import VectorAssembler
4. # 创建样本数据
5. data = spark.createDataFrame([
6.     (1.0, 1.0, 0.0),
7.     (1.0, 2.0, 1.0),
8.     (2.0, 2.0, 0.0),
9.     (2.0, 3.0, 1.0),
10. ], ["feature1", "feature2", "label"])
12. # 将特征组合成向量
13. assembler = VectorAssembler(inputCols=["feature1", "feature2"], outputCol="features")
14. data = assembler.transform(data)
16. # 创建决策树分类器
17. dt = DecisionTreeClassifier(featuresCol="features", labelCol="label")
19. # 训练模型
20. model = dt.fit(data)
22. # 预测
23. predictions = model.transform(data)
25. # 显示预测结果
26. predictions.select("features", "label", "prediction").show()

复制代码

更多功能、详细用法可参考官方文档：

1. https://spark.apache.org/docs/latest/api/python/index.html

复制代码

…
六、结语

今天我和各位靓仔们一起探索了 PySpark 这个强大的 Python 大数据处理库。从根本操作到高级应用，再到实际案例，信赖你已经对 PySpark 有了全面的了解和掌握。通过这种方式，我们可以更加轻松地处理大规模数据，提高数据处理的效率和结果。

渴望你能将这些知识应用到实际项目中，享受大数据处理的乐趣。编码不仅是一种技能，更是一门艺术。愿你在数据的海洋里不停探索，成为一个真正的数据大师！

…
七、作者Info

   Author：小鸿的摸鱼日常

Goal：让编程更风趣！ 专注于 Web 开发、爬虫，游戏开发，数据分析、天然语言处理，AI 等，期待你的关注，让我们一起发展、一起Coding！

版权说明：本文克制抄袭、转载，侵权必究！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。