PySpark：基于Reference DataFrame筛选后为Info DataFrame添加整列常量

需求说明 #

现有两个PySpark DataFrame：

• Info表：作为目标表，需要新增一列常量值
• Reference表：从中筛选key='b'的行，取其value值（即2）作为常量，添加到Info表中，生成指定结构的Output表

错误代码与报错信息 #

尝试以下代码时出现报错：

df_cal = (
    info
    .join(reference)
    .withColumn('const', reference.loc[reference['key']=='b', 'value'].iloc[0])
    .select('key', 'result', 'const')
)

df_cal.show()

报错信息：

AttributeError: 'Dataframe' object has no attribute 'loc'

各表结构 #

Info表：

+-----+-----+
|  key|value|
+-----+-----+
|    a|   10|
|    b|   20|
|    c|   50|
|    d|   40|
+-----+-----+

Reference表：

+-----+-----+
|  key|value|
+-----+-----+
|    a|   42|
|    b|    2|
|    c|    9|
|    d|  100|
+-----+-----+

期望Output表：

+-----+-----+-----+
|  key|value|const|
+-----+-----+-----+
|    a|   10|    2|
|    b|   20|    2|
|    c|   50|    2|
|    d|   40|    2|
+-----+-----+-----+

报错原因 #

你混淆了PySpark DataFrame和Pandas DataFrame的API：loc是Pandas用来定位行/列的方法，PySpark DataFrame并没有这个属性，因此抛出该错误。另外原代码还有两个问题：

• join(reference)未指定关联键，会生成笛卡尔积，完全没必要
• select中的result列在Info表中不存在，应该是value

解决方案 #

方法一：提取常量值后添加列（适合小表场景）

先从Reference表中提取目标常量值，再用lit函数将其转为PySpark列，添加到Info表中：

from pyspark.sql.functions import lit

# 从Reference中取出key='b'对应的value值
const_value = reference.filter(reference.key == 'b').select('value').collect()[0][0]

# 为Info表添加名为const的常量列
output_df = info.withColumn('const', lit(const_value))
output_df.show()

方法二：广播单行表+交叉关联（适合大表场景）

如果Reference表数据量很大，避免将数据拉到Driver节点，可以广播筛选后的单行表，再通过crossJoin关联：

from pyspark.sql.functions import broadcast

# 筛选key='b'的行，将value列重命名为const
const_df = reference.filter(reference.key == 'b').select('value').withColumnRenamed('value', 'const')

# 交叉关联Info表和广播后的常量表，实现常量列添加
output_df = info.crossJoin(broadcast(const_df))
output_df.show()

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