PySpark 创建嵌套的 DataFrame

在本文中，我们将介绍如何使用 PySpark 创建嵌套的 DataFrame。嵌套的 DataFrame 是一个包含有结构化列的列。这种数据结构可以让我们在一个列中存储更复杂的数据类型，比如数组、结构体和嵌套的表。

创建嵌套的 DataFrame

使用 PySpark 创建嵌套的 DataFrame 需要使用 StructType 类。StructType 是一个由 StructField 对象组成的列表，每个 StructField 对象包含列名和列的数据类型。以下是一个示例，展示了如何创建一个包含嵌套列的 DataFrame。

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, ArrayType

# 创建 SparkSession
spark = SparkSession.builder.getOrCreate()

# 定义 StructType 结构
schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", StringType(), True),
    StructField("address", StringType(), True),
    StructField("preference", ArrayType(StringType()), True)
])

# 创建 DataFrame
data = [("Alice", "25", "123 Main St", ["reading", "cooking"]),
        ("Bob", "30", "456 Oak Ave", ["music", "sports"])]

df = spark.createDataFrame(data, schema)

# 展示 DataFrame 结构
df.printSchema()
df.show()

在上面的示例中，我们首先导入了 SparkSession、StructType、StructField、StringType 和 ArrayType。然后我们创建了一个 SparkSession 对象 spark。接下来，我们定义了一个 StructType 结构，其中包含了四个字段：name、age、address 和 preference。preference 是一个包含多个偏好的数组字段。

之后，我们创建了一个数据列表 data，其中每个元组表示一行数据。接着，我们使用 spark.createDataFrame() 方法将数据列表 data 和结构类型 schema 传递给 createDataFrame 方法，从而创建了一个 DataFrame df。

最后，我们使用 df.printSchema() 方法打印 DataFrame 的结构，并使用 df.show() 方法显示 DataFrame 的内容。

操作嵌套的 DataFrame

一旦我们创建了嵌套的 DataFrame，我们可以使用 PySpark 提供的各种方法和函数来操作它。以下是一些常用的操作示例：

选择列

我们可以使用 select 方法选择特定的列。以下是一个示例：

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

过滤行

我们可以使用 filter 或 where 方法过滤 DataFrame 中的行。以下是一个示例：

df.filter(df.age > 25).show()

嵌套列的访问

我们可以使用点号（.）语法来访问嵌套列。以下是一个示例：

df.select("name", "preference[0]").show()

在上面的示例中，我们选择了 name 和 preference 列，同时访问了 preference 列中的第一个元素。

添加新列

我们可以使用 withColumn 方法向 DataFrame 中添加新列。以下是一个示例：

from pyspark.sql.functions import concat

df.withColumn("full_address", concat(df.address, lit(", "), df.city)).show()

在上面的示例中，我们使用 withColumn 方法创建了一个名为 full_address 的新列，该列通过拼接 address 和 city 列来创建。

更新嵌套列

我们可以使用 withColumn 方法更新嵌套列的值。以下是一个示例：

from pyspark.sql.functions import array_append

df.withColumn("preference", array_append(df.preference, "travel")).show()

在上面的例子中，我们使用 withColumn 方法将 “travel” 添加到 preference 列的末尾。

删除列

我们可以使用 drop 方法删除 DataFrame 中的列。以下是一个示例：

df.drop("address").show()

在上面的例子中，我们删除了 address 列。

总结

在本文中，我们介绍了如何使用 PySpark 创建嵌套的 DataFrame。我们了解了如何定义结构类型、创建嵌套列以及如何对嵌套的 DataFrame 进行各种操作，包括选择列、过滤行、访问嵌套列、添加新列、更新嵌套列和删除列。嵌套的 DataFrame 提供了一种更灵活的方式来组织和操作复杂的数据结构，使数据处理变得更加方便和高效。