PySpark实现合并同product_id的字典列表为唯一字典列表

问题分析 #

你需要将同一product_id对应的多个图片字典列表合并为一个平级列表，同时确保列表中的字典唯一。原有方案生成嵌套列表，是因为未正确拆解原列表中的字典元素，直接收集了原始列表而非单个字典。

解决方案 #

方法一：先拆解字典再收集去重 #

先将每个列表中的字典拆解为单独行，再按product_id分组收集唯一字典，生成平级列表：

from pyspark.sql import functions as F

# 拆解每个image列表中的字典，每行保留一个字典
exploded_df = df.select("product_id", F.explode("image").alias("image_item"))

# 按product_id分组，收集唯一字典组成最终列表
merged_df = exploded_df.groupBy("product_id").agg(F.collect_set("image_item").alias("image"))

方法二：先收集嵌套列表再展平去重 #

若不想拆解行，可先收集所有列表再展平，最后去重：

from pyspark.sql import functions as F

# 收集同一product_id的所有image列表，得到嵌套结构
collected_df = df.groupBy("product_id").agg(F.collect_list("image").alias("nested_image"))

# 展平嵌套列表并去重，得到平级的唯一字典列表
merged_df = collected_df.withColumn("image", F.array_distinct(F.flatten("nested_image")))

额外说明：若image列为字符串格式 #

如果你的image列是字符串类型（而非原生数组类型），需要先将字符串解析为数组再执行上述操作：

from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StructType, StringType, BooleanType

# 定义图片字典的结构
image_schema = ArrayType(StructType([
    F.StructField("url", StringType(), nullable=True),
    F.StructField("default", BooleanType(), nullable=True)
]))

# 将字符串格式的列表转换为Spark数组类型
df = df.withColumn("image", F.from_json("image", image_schema))

执行上述任一方法后，你将得到期望的输出：同一product_id对应一个平级的唯一字典列表。

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