如何实现两个PySpark DataFrame的逐行同列乘积计算

结构对齐说明 #

先明确两个表的匹配规则：

• df1 结构：首列为维度列product，其余为与df2完全同名的日期数值列
• df2 结构：仅1行数据，无维度列，列名与df1的日期列一一对应，存储的是对应列的乘法权重

本次实现不需要做表关联，直接逐列匹配同名列的权重值做乘法即可，全程无shuffle开销。

核心代码实现 #

1. 提取df2的权重映射
   因为df2只有单行数据，直接提取为字典结构，避免重复扫描全表：

# 生成格式为 {"04-01":1, "04-02":2, "04-03":3, "04-05":4, "04-06":5} 的权重字典
weight_dict = df2.first().asDict()

2. 构造计算逻辑
   遍历所有日期列，保留产品维度列不动，逐列生成乘法表达式：

from pyspark.sql.functions import col

# 初始化选列列表，先保留原始维度列
expr_list = [col("product")]
# 遍历权重字典的所有列名，生成乘法计算逻辑
for col_name in weight_dict:
    expr_list.append(
        (col(col_name) * weight_dict[col_name]).alias(col_name)
    )

3. 生成最终结果

result = df1.select(expr_list)
result.show()

常见坑点规避 #

• 前置做列名校验：如果df1与df2的日期列存在列名不匹配、列顺序错位的情况，需要先做列对齐，否则会触发KeyError
• 数据类型校验：样例中df1的列值存在带前导0的字符串格式（如03、04），需要先转换为整数/浮点型再计算，否则会出现类型运算报错
• 无需额外join操作：因为df2是全局单行权重，逐列乘法的写法性能远高于广播join实现，不会产生分布式计算的网络开销

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