Skip to main content

PySpark

PySpark 是 Apache Spark 的 Python API。Apache Spark 是一个强大的分布式数据处理框架，专用于大规模数据处理和数据分析。通过 PySpark，Python 开发者，即便不去深入了解分布式计算的复杂性，也可以方便地进行大规模数据分析和数据挖掘工作。

PySpark 主要包含以下这些组件：

  • Spark SQL：用于执行 SQL 查询以及读取数据的库，支持多种数据格式和存储系统。
  • DataFrame API：提供了一个分布式数据集合，使得数据处理和分析更加直观和高效。
  • MLlib：用于进行机器学习的库。
  • GraphX：用于图形处理的库（在 PySpark 中通过第三方库如 GraphFrames 访问）。
  • Spark Streaming：用于实时数据流处理的库。

其中数据库查询是最为常用的功能，所以我们这一章主要介绍数据库查询。

安装 PySpark​

要使用 PySpark，首先需要安装 Spark，并确保其在环境变量中正确配置。可以通过 pip 安装 PySpark：

pip install pyspark

安装后，我们可以用一个简单的 PySpark 示例程序测试一下：

# 导入 SparkSession，它是使用 Spark SQL 和 DataFrame 的入口点。
from pyspark.sql import SparkSession

# 初始化一个名为 “example” 的 Spark 会话。如果已经存在一个 Spark 会话，
# getOrCreate() 方法将返回现有的会话，否则，它将创建一个新的会话
spark = SparkSession.builder.appName("example").getOrCreate()

# 创建一个 DataFrame，其中包含两列（“ID” 和 “Value”）和三行数据。
# DataFrame 是分布式的数据集合，类似于关系数据库中的表。
df = spark.createDataFrame([
    (1, "foo"), 
    (2, "bar"), 
    (3, "baz")
], ["ID", "Value"])

# 将 DataFrame 的内容显示出来。默认情况下，show() 方法将显示 DataFrame 的前 20 行。
df.show()

连接到数据库​

使用 PySpark 肯定是为了操作数据库的，所以，安装之后的最重要步奏，就是连接数据库。PySpark 与数据库连接通常是通过 JDBC（Java Database Connectivity）完成的，JDBC 是一种用于执行 SQL 操作的 Java API，可以让你从 Spark 程序中访问几乎所有的关系型数据库。

Spark 本身是使用 Java 语言编写的，PySpark 只是为 Spark 包装了一层 Python 接口函数。所以在使用 PySpark 的时候，还是会调用到底层的 Java 库。

与数据库连接包括以下步骤：

1. 环境准备​

确保已经安装了 PySpark 并且配置了 Spark 环境。此外，还需要有目标数据库的 JDBC 驱动程序。例如，如果想连接到 MySQL 数据库，那么需要下载 MySQL 的 JDBC 驱动并将其放在类路径中。

2. SparkSession 创建​

SparkSession 是 Spark 2.0 以后引入的概念，它是程序执行的入口，用于配置 Spark 的各种设置（如 master URL）和初始化 Spark 应用。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DatabaseTableExample") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

3. 数据库连接​

要从 Spark 访问数据库，需要指定 JDBC 连接的 URL，用户名和密码（如果需要的话），以及要访问的表名。这可以通过 jdbc 方法完成，该方法是 DataFrameReader 的一部分。

# 数据库连接参数
jdbc_url = "jdbc:mysql://your_database_host:port/database_name"
connection_properties = {
    "user": "your_database_username",
    "password": "your_database_password",
    "driver": "com.mysql.jdbc.Driver"
}

# 读取数据库表
df = spark.read.jdbc(url=jdbc_url, table="your_table_name", properties=connection_properties)

4. 使用 DataFrame 操作数据​

一旦有了表示数据库表的 DataFrame，就可以使用 PySpark 提供的所有 DataFrame 操作来处理数据。例如，你可以进行筛选、聚合、连接等操作。

# 显示 DataFrame 的前几行
df.show()

# 进行简单的数据转换
filtered_df = df.filter(df["some_column"] > 100)

5. 关闭 SparkSession​

在程序的最后，不要忘记关闭 SparkSession。

spark.stop()

调用 PySpark 的两种方法​

通过 PySpark API 通过 Spark SQL 语句

各自优缺点

select​

使用 PySpark 选择（查询）表格中的数据是数据处理的基础步骤之一。PySpark 提供了灵活的 API 来查询和转换数据。以下是如何使用 PySpark 从 DataFrame 中选择数据的详细步骤，我们将以读取数据库表为例，然后展示如何对这些数据进行选择操作。

步骤 1: 初始化 SparkSession​

首先，你需要创建一个 SparkSession 对象，它是使用 PySpark 的入口。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("SelectDataExample") \
    .getOrCreate()

步骤 2: 读取数据​

这里假设你已经按照前面的指南将数据从数据库加载到 DataFrame 中。如果你是从其他数据源（如 CSV、JSON 文件或其他数据库）读取数据，方法类似，只是读取函数会有所不同（例如 spark.read.csv, spark.read.json 等）。

步骤 3: 选择数据​

3.1 选择特定的列​

使用 select 方法选择表中的特定列。以下示例展示如何选择名为 "name" 和 "age" 的列：

selected_data = df.select("name", "age")
selected_data.show()

3.2 使用表达式过滤数据​

你可以使用 filter 或 where 方法来基于条件过滤数据。这两个方法在功能上是相同的，可以互换使用。

filtered_data = df.filter(df["age"] > 30)
# 或者
filtered_data = df.where(df["age"] > 30)
filtered_data.show()

3.3 选择并转换数据​

你还可以在选择数据时对其进行转换。例如，增加一列，该列是现有列的转换结果。

from pyspark.sql import functions as F

transformed_data = df.select("name", "age", (F.col("age") + 10).alias("age_plus_ten"))
transformed_data.show()

步骤 4: 使用 SQL 查询数据​

另一种选择数据的方式是使用 SQL 语句。首先，你需要将 DataFrame 注册为临时视图。

df.createOrReplaceTempView("people")

然后，你可以使用 SQL 语句来查询数据。

sql_result = spark.sql("SELECT name, age FROM people WHERE age > 30")
sql_result.show()

步骤 5: 关闭 SparkSession​

在你的应用程序结束时，不要忘记关闭 SparkSession。

spark.stop()

总结​

使用 PySpark 选择数据非常灵活和强大，提供了多种方法来进行数据查询和转换。你可以根据需要使用 DataFrame API 或 SQL 语句来操作数据。通过这种方式，PySpark 使得处理大规模数据变得容易和高效。

JOIN​

在 PySpark 中，将多个表格（DataFrame）进行连接（join）是一项常用且强大的数据操作，它允许你将不同表格中的数据根据指定的条件合并起来。PySpark 支持多种类型的连接操作，包括内连接（inner join）、外连接（outer join，包括左外连接left outer join、右外连接right outer join）、交叉连接（cross join）等。

步骤 1: 初始化 SparkSession​

创建一个 SparkSession 对象，这是使用 PySpark 的第一步。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("JoinTablesExample") \
    .getOrCreate()

步骤 2: 创建或读取 DataFrame​

假设你已经有两个或多个要连接的 DataFrame。这些 DataFrame 可以是通过读取数据库、文件或其他数据源获得的，或者是直接在代码中创建的。

# 假设 df1 和 df2 是你要连接的两个 DataFrame
df1 = spark.createDataFrame([(1, "Alice"), (2, "Bob")], ["id", "name"])
df2 = spark.createDataFrame([(1, "New York"), (2, "Los Angeles")], ["id", "city"])

步骤 3: 进行连接操作​

3.1 内连接​

内连接返回两个表格中连接键匹配的行。

inner_join_df = df1.join(df2, df1.id == df2.id)
inner_join_df.show()

3.2 左外连接​

左外连接返回左表的所有行，即使在右表中没有匹配的行。如果右表中没有匹配，右表的列将返回null。

left_outer_join_df = df1.join(df2, df1.id == df2.id, "left_outer")
left_outer_join_df.show()

3.3 右外连接​

右外连接返回右表的所有行，即使在左表中没有匹配的行。

right_outer_join_df = df1.join(df2, df1.id == df2.id, "right_outer")
right_outer_join_df.show()

3.4 全外连接​

全外连接返回两个表格中所有行。如果某一侧没有匹配，则该侧的列将返回null。

full_outer_join_df = df1.join(df2, df1.id == df2.id, "full_outer")
full_outer_join_df.show()

3.5 交叉连接​

交叉连接返回两个表格的笛卡尔积，每个左表行与右表的每行组合。

cross_join_df = df1.crossJoin(df2)
cross_join_df.show()

步骤 4: 使用条件过滤​

连接操作还可以结合条件过滤使用，这通过在 .join() 方法中添加额外的筛选条件实现。

condition_join_df = df1.join(df2, (df1.id == df2.id) & (df1.name == "Alice"))
condition_join_df.show()

步骤 5: 关闭 SparkSession​

在应用程序结束时，关闭 SparkSession。

spark.stop()

注意事项​

  • 在进行连接操作时，注意处理可能的重复列名问题。一种常见的解决方案是使用别名（alias）为每个 DataFrame 的列提供唯一名称。
  • 当处理大数据集时，连接操作可能非常资源密集和时间消耗，尤其是交叉连接。务必优化你的查询和资源配置，以确保性能。
  • 使用适当的连接类型以避免不必要的数据丢失或冗余。明确你的数据关系和业务需求，选择最合适的连接类型。

通过上述步骤，你可以在 PySpark 中灵活地使用各种连接操作，有效地合并和分析跨多个表格的数据。

explode​

在 PySpark 中，explode 函数用于将数组类型的列展开成多行，这在处理嵌套数组或需要将单个复杂数据类型的列分解为多行数据时非常有用。以下是使用 PySpark explode 函数的详细介绍：

步骤 1: 初始化 SparkSession​

首先，创建一个 SparkSession 实例，它是使用 PySpark 进行任何操作的入口。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ExplodeExample") \
    .getOrCreate()

步骤 2: 创建或读取 DataFrame​

在这个例子中，假设我们有一个包含数组类型列的 DataFrame。如果你是从JSON文件、数据库或其他数据源读取数据，方法相似，只是读取函数会有所不同。

from pyspark.sql import Row
from pyspark.sql.functions import explode

# 创建一个包含数组类型列的 DataFrame
data = [Row(name="Alice", items=["apple", "banana", "orange"]),
        Row(name="Bob", items=["watermelon", "peach"])]
df = spark.createDataFrame(data)
df.show()

输出类似于：

+-----+-------------------+
| name|              items|
+-----+-------------------+
|Alice| [apple, banana, orange]|
| Bob|   [watermelon, peach]|
+-----+-------------------+

步骤 3: 使用 explode 函数​

explode 函数将每个数组元素转换为独立的行，复制数组外的其他列值。

from pyspark.sql.functions import explode

# 使用 explode 函数展开 items 列
exploded_df = df.select(df.name, explode(df.items).alias("item"))
exploded_df.show()

输出类似于：

+-----+----------+
| name|      item|
+-----+----------+
|Alice|     apple|
|Alice|    banana|
|Alice|    orange|
| Bob|watermelon|
| Bob|     peach|
+-----+----------+

步骤 4: 处理复杂的嵌套结构​

如果你的 DataFrame 包含更复杂的嵌套结构（例如，数组中嵌套的数组或结构体），你可以使用 explode 结合其他函数（如 col 或者 struct）来处理这些复杂的数据类型。

步骤 5: 关闭 SparkSession​

完成数据处理后，不要忘记关闭 SparkSession。

spark.stop()

注意事项​

  • 使用 explode 函数会显著增加 DataFrame 的行数，特别是当你展开的列包含大量元素时。这可能会影响性能，特别是在处理大数据集时。
  • 对于包含空数组或 null 值的行，explode 函数会将这些行从结果中删除。如果你希望保留这些行，可以考虑使用 explode_outer 函数，它会为空数组或 null 值生成一行，其中展开的列值为 null。
  • explode 可以与其他 DataFrame 操作结合使用，例如分组（groupBy）、聚合（agg）等，以实现更复杂的数据转换和分析任务。

通过上述步骤，你可以在 PySpark 中有效地使用 explode 函数来处理和分析嵌套数组类型的数据。

UDF​

............