如何用PySpark将DataFrame中多列分隔值对应展开并避免重复

你需要将ID1和ID2字段中的分隔值按对应关系展开为多行，避免重复记录，核心是保证两个字段的拆分元素一一匹配。以下是实现代码：

from pyspark.sql import functions as F

# 1. 清理ID1和ID2中的特殊字符（方括号、双引号）
df_clean = df.withColumn("ID1_clean", F.regexp_replace("ID1", r"\[|\]|\"", "")) \
             .withColumn("ID2_clean", F.regexp_replace("ID2", r"\[|\]|\"", ""))

# 2. 将清理后的字符串按逗号拆分为数组，对应原字段中逗号分隔的子项
df_arrays = df_clean.withColumn("ID1_array", F.split("ID1_clean", ",")) \
                   .withColumn("ID2_array", F.split("ID2_clean", ","))

# 3. 对数组内每个子项按"-"拆分，再扁平化得到最终的ID列表
df_flatten = df_arrays.withColumn("ID1_flat", F.flatten(F.transform("ID1_array", lambda x: F.split(x, "-")))) \
                     .withColumn("ID2_flat", F.flatten(F.transform("ID2_array", lambda x: F.split(x, "-"))))

# 4. 将两个ID列表按位置配对，再展开为多行
fulldf = df_flatten.withColumn("zipped", F.arrays_zip("ID1_flat", "ID2_flat")) \
                   .select("ID", F.explode("zipped").alias("pair")) \
                   .select("ID", "pair.ID1_flat".alias("ID1"), "pair.ID2_flat".alias("ID2"))

# 查看结果
fulldf.display()

步骤说明： #

• 步骤1：去除ID1、ID2字段中的[]"干扰字符，得到干净的字符串内容。
• 步骤2：将字符串按逗号拆分为数组，对应原数据中用逗号分隔的独立子项。
• 步骤3：用transform遍历数组内的每个子项，按-拆分为更小的数组，再通过flatten将嵌套数组转为一维数组，得到所有需要展开的ID集合。
• 步骤4：用arrays_zip将两个ID列表按位置一一配对，再通过explode将配对元素展开为单独行，确保ID1和ID2的元素完全对应，不会产生笛卡尔积式的重复记录。

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