PySpark DataFrame中如何展开ArrayType类型的Items与Value1列

我有如下PySpark DataFrame，需要展开其中的Items和Value1列。

当前使用的代码：

df_ob_exploded = df.withColumn('op_it.objects',
                    F.explode(F.array(*[F.array(F.col('op_it.objects')[i]['Items'][j].Name,
                                F.array(*[F.col('op_it.Objects')[i]['Items'[j]['Items'][k].Name for k in range(9)]).cast('string'),
                                F.array(*[F.from_json(F.col('op_it.objects')[i]['Items'][k].Selected, sc).getItem('Value') for k in range(9)]).cast('string'),
                                F.from_json(F.col('op_it.objects')[i]['Items'][j].Selected,sc).getItem('Value'))for i in range(9) for j in range(9)]))).dropDuplicates()
                                
df_obj_json = df_ob_exploded.select('ID', F.col('op_it.object'[0].alias('Name'), F.col('op_it.objects')[1].alias('Items'),
                                          F.col('op_it.object'[2].alias('Value1'), F.col('op_it.objects')[1].alias('Value'))\
                                          .na.drop(how='all', subset=['Name','Items','Value1','Value'])

但无法成功展开Items和Value1列，我的DataFrame示例如下：

ID      Name        Items                   Value1      Value
1       Contact     [platform,chat, , ,]    [,,,,,]     null
1       action      [,,,,,,]                [,,,,,]     windows
1       cycle       [,,,,,,]                [,,,,,]     article

要同时展开Items和Value1数组列且保持元素对应关系，不需要复杂的硬编码遍历，用arrays_zip打包数组后再explode即可，步骤如下：

1. 打包并展开数组 #

用arrays_zip将Items和Value1中同位置的元素配对成结构体数组，再通过explode拆分成单独行：

from pyspark.sql import functions as F

# 打包数组并展开
df_exploded = df.withColumn(
    "zipped",
    F.explode(F.arrays_zip("Items", "Value1"))
).select(
    "ID", "Name", "Value",
    F.col("zipped.Items").alias("Item"),
    F.col("zipped.Value1").alias("Value1_element")
)

2. 过滤无效空值 #

原数组包含大量空字符串，过滤掉这些无效行：

df_result = df_exploded.filter(
    (F.col("Item") != "") & (F.col("Value1_element") != "")
)

完整测试示例 #

基于你提供的测试数据运行完整代码：

# 创建测试DataFrame
data = [
    (1, "Contact", ["platform", "chat", "", "", ""], ["", "", "", "", ""], None),
    (1, "action", ["", "", "", "", "", ""], ["", "", "", "", "", ""], "windows"),
    (1, "cycle", ["", "", "", "", "", ""], ["", "", "", "", "", ""], "article")
]
df = spark.createDataFrame(data, ["ID", "Name", "Items", "Value1", "Value"])

# 执行展开和过滤
df_exploded = df.withColumn(
    "zipped",
    F.explode(F.arrays_zip("Items", "Value1"))
).select(
    "ID", "Name", "Value",
    F.col("zipped.Items").alias("Item"),
    F.col("zipped.Value1").alias("Value1_element")
)

df_result = df_exploded.filter(
    (F.col("Item") != "") & (F.col("Value1_element") != "")
)

df_result.show()

关键说明 #

• arrays_zip确保展开后Item和Value1_element始终是原数组中同一位置的对应元素，避免错位
• 无需硬编码数组长度，适配任意长度的数组列
• 过滤条件可根据实际需求调整，比如只保留非空的Item或Value1_element

内容的提问来源于stack exchange，提问作者user175025