PySpark 将字符串化字典数组展开为行

在本文中，我们将介绍如何在PySpark中将字符串化的字典数组展开为独立的行。PySpark是Apache Spark的Python API，是一个开源的大数据处理框架，适用于处理大规模数据集。

字符串化的字典数组与展开

在处理大规模数据集时，经常会遇到包含复杂数据类型的列。例如，在一个数据集中，可能会有一个包含字典的数组类型的列。而有时候，这个数组会被字符串化存储。在这种情况下，我们需要将字符串化的字典数组展开为独立的行，以便更容易进行数据分析和处理。

假设我们有以下字符串化的字典数组列：

id	data
1	‘[{“name”: “Alice”, “age”: “25”}, {“name”: “Bob”, “age”: “30”}]’
2	‘[{“name”: “Charlie”, “age”: “35”}, {“name”: “David”, “age”: “40”}]’
3	‘[{“name”: “Eve”, “age”: “45”}, {“name”: “Frank”, “age”: “50”}]’

创建DataFrame

首先，我们需要创建一个DataFrame来处理数据。我们可以使用SparkSession创建一个DataFrame，并将数据加载到其中。

# 导入必要的模块
from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder \
    .appName("Explode Example") \
    .getOrCreate()

# 创建数据集
data = [(1, '[{"name": "Alice", "age": "25"}, {"name": "Bob", "age": "30"}]'),
        (2, '[{"name": "Charlie", "age": "35"}, {"name": "David", "age": "40"}]'),
        (3, '[{"name": "Eve", "age": "45"}, {"name": "Frank", "age": "50"}]')]

# 创建DataFrame
df = spark.createDataFrame(data, ["id", "data"])

现在，我们已经创建了一个包含”id”和”data”列的DataFrame。”data”列包含了字符串化的字典数组。接下来，我们将展示如何将字符串化的字典数组展开为独立的行。

使用explode和from_json函数进行展开

在PySpark中，我们可以使用explode函数和from_json函数来展开字符串化的字典数组。首先，我们需要将字符串化的字典数组转换为结构化的DataFrame，然后使用explode函数展开数组中的每个元素。

# 导入必要的函数
from pyspark.sql.functions import explode, from_json

# 使用explode函数展开字符串化的字典数组
df_exploded = df.withColumn("exploded_data", explode(from_json(df.data, "array<struct<name:string, age:string>>")))

# 展示结果
df_exploded.show()

运行上述代码后，我们可以看到展开前后的结果：

id	data	exploded_data
1	‘[{“name”: “Alice”, “age”: “25”}, {“name”: “Bob”, “age”: “30”}]’	{Alice, 25}
1	‘[{“name”: “Alice”, “age”: “25”}, {“name”: “Bob”, “age”: “30”}]’	{Bob, 30}
2	‘[{“name”: “Charlie”, “age”: “35”}, {“name”: “David”, “age”: “40”}]’	{Charlie, 35}
2	‘[{“name”: “Charlie”, “age”: “35”}, {“name”: “David”, “age”: “40”}]’	{David, 40}
3	‘[{“name”: “Eve”, “age”: “45”}, {“name”: “Frank”, “age”: “50”}]’	{Eve, 45}
3	‘[{“name”: “Eve”, “age”: “45”}, {“name”: “Frank”, “age”: “50”}]’	{Frank, 50}

我们可以看到，原来的DataFrame中的每个数组元素被展开为一个独立的行。展开后的结果保存在新的列”exploded_data”中。

提取展开后的列

在展开后，我们可以使用select函数来提取所需的列。可以根据需要选择原始的”id”和”data”列，以及展开后的”exploded_data”列。

# 提取所需的列
df_extracted = df_exploded.select("id", "data", "exploded_data")

# 展示结果
df_extracted.show()

运行上述代码后，我们可以看到提取后的结果：

id	data	exploded_data
1	‘[{“name”: “Alice”, “age”: “25”}, {“name”: “Bob”, “age”: “30”}]’	{Alice, 25}
1	‘[{“name”: “Alice”, “age”: “25”}, {“name”: “Bob”, “age”: “30”}]’	{Bob, 30}
2	‘[{“name”: “Charlie”, “age”: “35”}, {“name”: “David”, “age”: “40”}]’	{Charlie, 35}
2	‘[{“name”: “Charlie”, “age”: “35”}, {“name”: “David”, “age”: “40”}]’	{David, 40}
3	‘[{“name”: “Eve”, “age”: “45”}, {“name”: “Frank”, “age”: “50”}]’	{Eve, 45}
3	‘[{“name”: “Eve”, “age”: “45”}, {“name”: “Frank”, “age”: “50”}]’	{Frank, 50}