数据结构
网络
RDBMS
操作系统
Java
MS Excel
iOS
HTML
CSS
Android
Python
C 编程
C++
C#
MongoDB
MySQL
Javascript
PHP
物理学
化学
生物学
数学
英语
经济学
心理学
社会学
服装设计
法律研究使用 Python PySpark 处理大型数据集
在本教程中，我们将探索 Python 和 PySpark 的强大组合，用于处理大型数据集。PySpark 是一个 Python 库，它为 Apache Spark 提供了一个接口，Apache Spark 是一个快速且通用的集群计算系统。通过利用 PySpark，我们可以有效地将数据分布和处理到多个机器的集群中，使我们能够轻松处理大规模数据集。
在本文中，我们将深入探讨 PySpark 的基础知识，并演示如何在大型数据集上执行各种数据处理任务。我们将涵盖关键概念，例如 RDD（弹性分布式数据集）和 DataFrame，并通过分步示例展示它们的实际应用。在本教程结束时，您将对如何利用 PySpark 有效地处理和分析海量数据集有一个扎实的了解。
第 1 部分：PySpark 入门
在本节中，我们将设置我们的开发环境，并熟悉 PySpark 的基本概念。我们将介绍如何安装 PySpark、初始化 SparkSession 以及将数据加载到 RDD 和 DataFrame 中。让我们从安装 PySpark 开始
# Install PySpark !pip install pyspark
输出
Collecting pyspark ... Successfully installed pyspark-3.1.2
安装 PySpark 后，我们可以初始化一个 SparkSession 以连接到我们的 Spark 集群
from pyspark.sql import SparkSession
# Create a SparkSession
spark = SparkSession.builder.appName("LargeDatasetProcessing").getOrCreate()
准备好 SparkSession 后，我们现在可以将数据加载到 RDD 或 DataFrame 中。RDD 是 PySpark 中的基本数据结构，并提供了一个分布式元素集合。另一方面，DataFrame 将数据组织成命名列，类似于关系数据库中的表。让我们将 CSV 文件加载为 DataFrame
# Load a CSV file as a DataFrame
df = spark.read.csv("large_dataset.csv", header=True, inferSchema=True)
输出
+---+------+--------+ |id |name |age | +---+------+--------+ |1 |John |32 | |2 |Alice |28 | |3 |Bob |35 | +---+------+--------+
从上面的代码片段中可以看到，我们使用 `read.csv()` 方法将 CSV 文件读取到数据框中。`header=True` 参数表示第一行包含列名，`inferSchema=True` 自动推断每列的数据类型。
第 2 部分：数据转换和分析
在本节中，我们将探索使用 PySpark 的各种数据转换和分析技术。我们将介绍诸如过滤、聚合和连接数据集等操作。让我们从根据特定条件过滤数据开始
# Filter data filtered_data = df.filter(df["age"] > 30)
输出
+---+----+---+ |id |name|age| +---+----+---+ |1 |John|32 | |3 |Bob |35 | +---+----+---+
在上面的代码片段中，我们使用 `filter()` 方法选择“age”列大于 30 的行。此操作允许我们从大型数据集中提取相关的数据子集。
接下来，让我们使用 `groupBy()` 和 `agg()` 方法对数据集执行聚合
# Aggregate data
aggregated_data = df.groupBy("gender").agg({"salary": "mean", "age": "max"})
输出
+------+-----------+--------+ |gender|avg(salary)|max(age)| +------+-----------+--------+ |Male |2500 |32 | |Female|3000 |35 | +------+-----------+--------+
在这里，我们按“gender”列对数据进行分组，并计算每个组的平均工资和最大年龄。生成的 `aggregated_data` DataFrame 为我们提供了对数据集的有价值的见解。
除了过滤和聚合之外，PySpark 还使我们能够有效地连接多个数据集。让我们考虑一个我们有两个 DataFrame 的示例：`df1` 和 `df2`。我们可以根据公共列连接它们
# Join two DataFrames joined_data = df1.join(df2, on="id", how="inner")
输出
+---+----+---------+------+ |id |name|department|salary| +---+----+---------+------+ |1 |John|HR |2500 | |2 |Alice|IT |3000 | |3 |Bob |Sales |2000 | +---+----+---------+------+
`join()` 方法允许我们根据由 `on` 参数指定的公共列组合 DataFrame。我们可以根据需要选择不同的连接类型，例如“inner”、“outer”、“left”或“right”。
第 3 部分：高级 PySpark 技术
在本节中，我们将探索高级 PySpark 技术以进一步增强我们的数据处理能力。我们将介绍用户定义函数 (UDF)、窗口函数和缓存等主题。让我们从定义和使用 UDF 开始
from pyspark.sql.functions import udf
# Define a UDF
def square(x):
return x ** 2
# Register the UDF
square_udf = udf(square)
# Apply the UDF to a column
df = df.withColumn("age_squared", square_udf(df["age"]))
输出
+---+------+---+------------+ |id |name |age|age_squared | +---+------+---+------------+ |1 |John |32 |1024 | |2 |Alice |28 |784 | |3 |Bob |35 |1225 | +---+------+---+------------+
在上面的代码片段中，我们定义了一个名为 `square()` 的简单 UDF，它对给定的输入进行平方。然后，我们使用 `udf()` 函数注册 UDF 并将其应用于“age”列，在我们的 DataFrame 中创建一个名为“age_squared”的新列。
PySpark 还提供了强大的窗口函数，使我们能够在特定窗口范围内执行计算。让我们计算每个员工的平均工资，同时考虑前一行和下一行
from pyspark.sql.window import Window
from pyspark.sql.functions import lag, lead, avg
# Define the window
window = Window.orderBy("id")
# Calculate average salary with lag and lead
df = df.withColumn("avg_salary", (lag(df["salary"]).over(window) + lead(df["salary"]).over(window) + df["salary"]) / 3)
输出
+---+----+---------+------+----------+ |id |name|department|salary|avg_salary| +---+----+---------+------+----------+ |1 |John|HR |2500 |2666.6667 | |2 |Alice| IT |3000 |2833.3333 | |3 |Bob |Sales |2000 |2500 | +---+----+---------+------+----------+
在上面的代码片段中，我们使用 `Window.orderBy()` 方法定义了一个窗口，指定基于“id”列的行排序。然后，我们分别使用 `lag()` 和 `lead()` 函数访问前一行和下一行。最后，我们通过考虑当前行及其邻居来计算平均工资。
最后，缓存是 PySpark 中一项重要的技术，用于提高迭代算法或重复计算的性能。我们可以使用 `cache()` 方法将 DataFrame 或 RDD 缓存到内存中
# Cache a DataFrame df.cache()
缓存不会显示任何输出，但依赖于缓存 DataFrame 的后续操作将更快，因为数据存储在内存中。
结论
在本教程中，我们探索了 PySpark 在 Python 中处理大型数据集的强大功能。我们首先设置了我们的开发环境并将数据加载到 RDD 和 DataFrame 中。然后，我们深入研究了数据转换和分析技术，包括过滤、聚合和连接数据集。最后，我们讨论了高级 PySpark 技术，例如用户定义函数、窗口函数和缓存。
1K+ 次查看
