PySpark DataFrame左连接后列值更新不符合预期问题求助

问题背景 #

对两个PySpark DataFrame：customer和price执行左连接，连接条件为price.PRICE_CODE == customer.C_CODE 且 price.PRODUCT_LOCATION == customer.CUSTOMER_LOCATION，但派生列derived在无匹配记录时未正确赋值为PRICE_DEFAULT的平均值。

数据Schema #

customer表Schema #

customer:
 root
 |-- NAME: string (nullable = true)
 |-- C_CODE: string (nullable = true)
 |-- C_OPTION: string (nullable = true)
 |-- C_MATERIAL: string(10,0) (nullable = true)
 |-- CID: string (nullable = true)
 |-- CUSTOMER_EXPENSES: string (nullable = true)
 |-- CUSTOMER_LOCATION: string (nullable = true)
 |-- PRODUCT_NAME: string (nullable = true)

price表Schema #

price:
 root
 |-- PRICE_ID: string (nullable = true)
 |-- PRICE_CODE: string (nullable = true)
 |-- PRICE_RANGE: string (nullable = true)
 |-- C_MATERIAL: string(10,0) (nullable = true)
 |-- CID: string (nullable = true)
 |-- PRICE_DEFAULT: int (nullable = true)
 |-- PRODUCT_LOCATION: string (nullable = true)
 |-- PRODUCT_NAME: string (nullable = true)

需求逻辑 #

• 匹配到对应记录时，derived取值为customer.CUSTOMER_EXPENSES
• 无匹配记录时，derived取值为price表中PRICE_DEFAULT列的平均值

当前实现代码 #

左连接代码 #

join_df = price.join(customer, on=(price['PRICE_CODE'] == customer['C_CODE']) & (price['PRODUCT_LOCATION'] == customer['CUSTOMER_LOCATION']), how='left')

计算平均值 #

price_avg = str(join_df.select(avg('PRICE_DEFAULT')).collect()[0][0])

列值更新代码 #

price.join(customer, on=(price['PRICE_CODE'] == customer['C_CODE']) & (price['PRODUCT_LOCATION'] == customer['CUSTOMER_LOCATION']), how='left')
.drop(price.PRODUCT_NAME)\
.withColumn('derived', 
   when((col('PRICE_CODE').isNotNull()) & (col('PRODUCT_LOCATION').isNotNull()), customer.CUSTOMER_EXPENSES).\
   when((col('C_CODE').isNull()) & (col('CUSTOMER_LOCATION').isNull()), price_avg)
)

问题现象 #

匹配记录的derived列值符合预期，但无匹配记录（即customer表的C_CODE和CUSTOMER_LOCATION为NULL）时，derived列未显示平均值，仍保留customer.CUSTOMER_EXPENSES的值。

错误排查与修正方案 #

错误点分析 #

1. 条件判断逻辑错误：第一个when判断的是左表price的字段非空，左连接时左表字段不会因无匹配变为NULL，导致该条件永远为真，第二个when无法触发。应判断右表customer的字段是否为空（无匹配的标志）。
2. 平均值计算错误：从join_df计算平均值会包含左连接后的NULL值，导致平均值不准确；同时将平均值转为字符串会引发类型不匹配问题。
3. 代码冗余：重复执行左连接操作，未复用已生成的join_df。

修正后的完整代码 #

from pyspark.sql.functions import avg, when, col

# 直接从原price表计算PRICE_DEFAULT的平均值，避免左连接结果干扰
price_avg = price.select(avg('PRICE_DEFAULT')).collect()[0][0]

# 执行左连接并清理冗余字段
join_df = price.join(
    customer,
    on=(price['PRICE_CODE'] == customer['C_CODE']) & (price['PRODUCT_LOCATION'] == customer['CUSTOMER_LOCATION']),
    how='left'
).drop(price.PRODUCT_NAME)

# 正确生成derived列
final_df = join_df.withColumn(
    'derived',
    when(
        # 右表字段非空，说明存在匹配记录
        col('C_CODE').isNotNull() & col('CUSTOMER_LOCATION').isNotNull(),
        col('CUSTOMER_EXPENSES')
    ).otherwise(
        # 无匹配时使用预计算的平均值
        price_avg
    )
)

修正说明 #

• 调整匹配判断逻辑：通过右表C_CODE和CUSTOMER_LOCATION是否非空识别匹配记录，左连接无匹配时右表字段会自动变为NULL。
• 修正平均值计算：直接从原price表计算，保证平均值是全表真实平均，不受连接结果影响。
• 简化分支逻辑：用otherwise统一处理所有无匹配场景，代码更简洁清晰。
• 保留数值类型：避免将平均值转为字符串，防止与CUSTOMER_EXPENSES的类型冲突（若需统一类型，可按需转换）。

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