Spark DataFrame追加列问题：循环迭代时避免覆盖原有列

问题描述 #

现有如下PySpark代码，意图遍历impcomp列表中的每个字段，将处理后的列追加到主DataFrame dfu，但当前代码每次循环都会覆盖之前的列，最终仅保留最后一个字段unit：

impcomp = ['connectors', 'contract_no', 'document_confidentiality', 'document_type', 'drawing_no', 'equipment_specifications', 'external_drawings', 'is_psi', 'line_numbers', 'owner_no', 'plant', 'project_title', 'psi_category', 'revision', 'revision_date', 'revision_status', 'tags', 'unit']


for el in impcomp:
    df1 = df3.select(df[Pk[1]],posexplode_outer(df3[el]))
    df1 = df1.where(df1.pos !='1')
    df1 = df1.drop('pos')
    df1 = df1.withColumnRenamed('col',el)
    dfu = df4.join(df1,df4.DocumentNo ==  df1.DocumentNo,"left")

display(dfu)

尝试了另一段代码，触发错误AttributeError: 'list' object has no attribute '_jdf'：

dfs = []
for el in impcomp:
    df1 = df3.select(df[Pk[1]],posexplode_outer(df3[el]))
    df1 = df1.where(df1.pos !='1')
    df1 = df1.drop('pos')
    df1 = df1.withColumnRenamed('col',el)
    dfs.append(df1[el])
df6 = reduce(df4.union(dfs))

解决方案 #

核心问题分析 #

• 原始代码每次循环都用原始的df4与处理后的df1做join，导致每次都会覆盖之前的结果，仅保留当前循环的列
• 尝试的错误代码误用了union（用于行合并），且添加到列表的是Column对象而非DataFrame，reduce用法也不正确

正确实现代码 #

from pyspark.sql.functions import posexplode_outer

impcomp = ['connectors', 'contract_no', 'document_confidentiality', 'document_type', 'drawing_no', 'equipment_specifications', 'external_drawings', 'is_psi', 'line_numbers', 'owner_no', 'plant', 'project_title', 'psi_category', 'revision', 'revision_date', 'revision_status', 'tags', 'unit']

# 初始化dfu为主DataFrame df4
dfu = df4

for el in impcomp:
    # 处理当前字段，生成包含关联键DocumentNo和目标列的临时DataFrame
    df1 = df3.select(df[Pk[1]], posexplode_outer(df3[el]))
    # 注意：如果pos是整数类型，需改为df1.pos != 1（而非字符串'1'）
    df1 = df1.where(df1.pos != '1')
    df1 = df1.drop('pos')
    df1 = df1.withColumnRenamed('col', el)
    
    # 基于当前已有的dfu进行左join，保留所有已追加的列并添加新列
    dfu = dfu.join(df1, on=['DocumentNo'], how='left')

display(dfu)

关键说明 #

1. 初始化dfu：将dfu初始化为主DataFrame df4，后续每次循环都在最新的dfu基础上进行join，确保之前的列不会丢失
2. join逻辑优化：使用on=['DocumentNo']指定关联键，避免生成重复的DocumentNo列
3. 数据类型注意：如果pos字段是整数类型，df1.pos != '1'会因为类型不匹配导致过滤失效，需改为df1.pos != 1

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