将硬编码的PySpark代码改写为循环实现（适配多列场景）

实现思路 #

通过循环遍历每个因子编号（从1到因子列的总数），动态生成关联条件、列重命名规则和冗余列清理逻辑，逐步构建最终的结果DataFrame。

具体代码实现 #

from pyspark.sql import functions as F

# 初始化结果DataFrame为原始保单表
policies_coeff = test_policies_rl

# 动态获取rates_table中rating_factor_*列的数量，确定循环范围
rating_factor_cols = [col for col in rates_table.columns if col.startswith("rating_factor_")]
num_factors = len(rating_factor_cols)

# 循环处理每个因子编号n（从1到因子总数）
for n in range(1, num_factors + 1):
    # 构建动态关联条件
    join_condition = (
        (policies_coeff.version == rates_table.version) &
        (policies_coeff.sub_class == rates_table.cover) &
        (policies_coeff[f"rf_{n}"] == rates_table.RF) &
        (policies_coeff[f"rl_{n}"] == rates_table[f"rating_factor_{n}"])
    )
    
    # 执行左关联 + 列重命名 + 清理冗余列
    policies_coeff = (
        policies_coeff.join(rates_table, join_condition, "left")
        .withColumnRenamed("rating_factor_amount", f"rating_factor_amount_{n}")
        .withColumnRenamed("rating_factor_coefficient", f"rating_factor_coefficient_{n}")
        .withColumnRenamed("rating_factor_level", f"rating_factor_level_{n}")
        .withColumnRenamed("rating_factor_rate", f"rating_factor_rate_{n}")
        .drop(
            rates_table.version, rates_table.cover, rates_table.RF,
            rates_table[f"rating_factor_{n}"]
        )
    )

关键细节说明 #

• 动态列名处理：用f-string直接拼接列名，PySpark支持通过df[列名字符串]的方式引用列，无需额外转换。
• 循环累积操作：每次循环都在之前的结果DataFrame基础上继续关联，确保所有因子的计算结果都被整合。
• 冗余列清理：每次关联后立即删除rates_table带来的重复列，避免后续循环出现列名冲突。

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