PySpark DataFrame聚合计算异常求助：过滤后结果不符预期

你遇到的问题核心在于聚合操作的上下文错误——虽然你筛选出了filter_df，但最终的聚合计算实际上是跑在原始的df上，而非筛选后的数据集。

为什么会出现这个错误？ #

当你写下：

result = spark_sum(filter_df['_2'] * filter_df['_3']) / spark_sum(filter_df['_3'])
df.select(result).show()

这里的spark_sum(filter_df['_2'] * filter_df['_3'])只是定义了一个聚合表达式，但这个表达式并没有和filter_df的执行逻辑绑定。当你把它传入df.select()时，Spark会默认以原始df作为计算上下文，忽略filter_df的筛选条件。

你可以验证一下：原始df中所有行的_2*_3总和是1*1 +1*0.2 +1*1 +1*0.5 +0.55*1 +1*1 = 4.25，_3的总和是1+0.2+1+0.5+1+1=4.7，4.25/4.7≈0.904，正好是你得到的错误结果。

正确的解决方法 #

要让聚合计算基于筛选后的filter_df，你需要在filter_df的上下文里执行聚合操作，推荐用agg()方法：

方法1：分步实现（更清晰）

from pyspark.sql.functions import sum as spark_sum

# 1. 筛选符合条件的行
filter_df = df.where(df['_2'] > df['_3'])

# 2. 在筛选后的数据集上执行聚合计算
result_df = filter_df.agg(
    (spark_sum("_2" * "_3") / spark_sum("_3")).alias("calculated_result")
)

# 3. 查看结果
result_df.show()

方法2：合并成单条语句（更简洁）

from pyspark.sql.functions import sum as spark_sum

result_df = df.where(df['_2'] > df['_3']).agg(
    (spark_sum("_2" * "_3") / spark_sum("_3")).alias("calculated_result")
)

result_df.show()

运行以上任意一种方法，都会得到你预期的结果1.0，因为聚合操作完全基于筛选后的两行数据：(1.0*0.2 +1.0*0.5)/(0.2+0.5) = 0.7/0.7=1.0。

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