使用Python PySpark处理大型数据集

百变鹏仔 1个月前 (01-20) #Python

文章标签数据

在本教程中，我们将探索Python和PySpark的强大组合，用于处理大型数据集。PySpark是一个Python库，提供了与Apache Spark的接口，它是一个快速且通用的集群计算系统。通过利用PySpark，我们可以高效地在一组机器上分发和处理数据，使我们能够轻松处理大规模数据集。

在本文中，我们将深入探讨PySpark的基本原理，并演示如何在大型数据集上执行各种数据处理任务。我们将涵盖关键概念，如RDD（弹性分布式数据集）和数据框架，并通过逐步示例展示它们的实际应用。通过本教程的学习，您将对如何有效地利用PySpark处理和分析大规模数据集有一个扎实的理解。

Section 1: Getting Started with PySpark

的中文翻译为：

第一部分：开始使用PySpark

在本节中，我们将设置开发环境并熟悉PySpark的基本概念。我们将介绍如何安装PySpark，初始化SparkSession，并将数据加载到RDD和DataFrame中。让我们开始安装PySpark：

# Install PySpark!pip install pyspark

输出

Collecting pyspark...Successfully installed pyspark-3.1.2

安装PySpark之后，我们可以初始化一个SparkSession来连接到我们的Spark集群：

立即学习“Python免费学习笔记（深入）”；

from pyspark.sql import SparkSession# Create a SparkSessionspark = SparkSession.builder.appName("LargeDatasetProcessing").getOrCreate()

有了我们准备好的SparkSession，我们现在可以将数据加载到RDDs或DataFrames中。RDDs是PySpark中的基本数据结构，它提供了一个分布式的元素集合。而DataFrames则将数据组织成命名列，类似于关系数据库中的表格。让我们将一个CSV文件加载为一个DataFrame：

# Load a CSV file as a DataFramedf = 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 datafiltered_data = df.filter(df["age"] &gt; 30)

输出

+---+----+---+|id |name|age|+---+----+---+|1  |John|32 ||3  |Bob |35 |+---+----+---+

在上面的代码片段中，我们使用`filter()`方法来选择“age”列大于30的行。这个操作允许我们从大型数据集中提取相关的子集。

接下来，让我们使用“groupBy()”和“agg()”方法对数据集执行聚合：

# Aggregate dataaggregated_data = df.groupBy("gender").agg({"salary": "mean", "age": "max"})

输出

+------+-----------+--------+|gender|avg(salary)|max(age)|+------+-----------+--------+|Male  |2500       |32      ||Female|3000       |35      |+------+-----------+--------+

在这里，我们按“性别”列对数据进行分组，并计算每组的平均工资和最大年龄。生成的“aggreated_data”数据框架为我们提供了对数据集的宝贵见解。

除了过滤和聚合之外，PySpark 还使我们能够高效地连接多个数据集。让我们考虑一个示例，其中我们有两个 DataFrame：“df1”和“df2”。我们可以根据一个共同的列加入它们：

# Join two DataFramesjoined_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"。

第三部分：高级PySpark技术

在本节中，我们将探讨高级的PySpark技术，以进一步增强我们的数据处理能力。我们将涵盖用户定义函数（UDFs）、窗口函数和缓存等主题。让我们从定义和使用UDF开始：

from pyspark.sql.functions import udf# Define a UDFdef square(x):    return x ** 2# Register the UDFsquare_udf = udf(square)# Apply the UDF to a columndf = 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        |+---+------+---+------------+

在上面的代码片段中，我们定义了一个简单的UDF函数，名为`square()`，它用于对给定的输入进行平方运算。然后，我们使用`udf()`函数注册该UDF，并将其应用于"age"列，从而在我们的DataFrame中创建一个名为"age_squared"的新列。

PySpark还提供了强大的窗口函数，允许我们在特定的窗口范围内执行计算。让我们考虑上一行和下一行来计算每个员工的平均工资：

from pyspark.sql.window import Windowfrom pyspark.sql.functions import lag, lead, avg# Define the windowwindow = Window.orderBy("id")# Calculate average salary with lag and leaddf = 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 DataFramedf.cache()

缓存不会显示任何输出，但依赖缓存的 DataFrame 的后续操作会更快，因为数据存储在内存中。

结论

在本教程中，我们探索了 PySpark 在 Python 中处理大型数据集的强大功能。我们首先设置开发环境并将数据加载到 RDD 和 DataFrame 中。然后，我们深入研究了数据转换和分析技术，包括过滤、聚合和连接数据集。最后，我们讨论了高级 PySpark 技术，例如用户定义函数、窗口函数和缓存。

文章推荐

使用Python PySpark处理大型数据集

Section 1: Getting Started with PySpark

第一部分：开始使用PySpark

输出

输出

第 2 部分：转换和分析数据

输出

输出

输出

第三部分：高级PySpark技术

输出

输出

结论

Python实现字典的key和values的交换

使用Python脚本来获取Cisco设备信息的示例

Python的Django中django-userena组件的简单使用教程

零基础写python爬虫之神器正则表达式

零基础写python爬虫之抓取百度贴吧代码分享