PySpark DataFrame分组转JSON时去重合并多联系信息问题咨询

出现重复的核心原因是collect_list会收集分组内所有行的对应字段值，未对重复的联系信息条目做过滤，可按以下两种方案实现：

方案1：分组前提前去重（推荐，性能更优） #

先按客户唯一标识+联系信息全字段做全局去重，再分组聚合，避免分组时处理冗余数据。

from pyspark.sql import functions as F

# 1. 去重：参数为客户唯一标识 + 地址全字段 + 邮箱全字段 + 电话全字段，字段名请替换为你的实际列名
df_dedup = df.dropDuplicates([
    "CustId", "TIN", "EntityType", "EntityAttributes",
    "address_line1", "address_line2", "city", "postcode", "address_type",
    "email_addr", "email_type",
    "phone_num", "phone_type"
])

# 2. 分组聚合生成嵌套结构
result_df = df_dedup.groupBy("CustId", "TIN", "EntityType", "EntityAttributes") \
    .agg(
        F.collect_list(F.struct(
            "address_line1", "address_line2", "city", "postcode", "address_type"
        )).alias("address_list"),
        F.collect_list(F.struct("email_addr", "email_type")).alias("email_list"),
        F.collect_list(F.struct("phone_num", "phone_type")).alias("phone_list")
    )

# 3. 输出JSON
# 按行输出多个JSON文件
result_df.write.mode("overwrite").json("./customer_json_output/")
# 输出为单个数组格式JSON文件
result_df.toPandas().to_json("./customer_single.json", orient="records", force_ascii=False)

方案2：聚合后对数组去重 #

如果不需要修改原有分组逻辑，可直接用array_distinct函数对collect_list生成的数组做去重：

from pyspark.sql import functions as F

result_df = df.groupBy("CustId", "TIN", "EntityType", "EntityAttributes") \
    .agg(
        F.array_distinct(F.collect_list(F.struct(
            "address_line1", "address_line2", "city", "postcode", "address_type"
        ))).alias("address_list"),
        F.array_distinct(F.collect_list(F.struct("email_addr", "email_type"))).alias("email_list"),
        F.array_distinct(F.collect_list(F.struct("phone_num", "phone_type"))).alias("phone_list")
    )

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