PySpark 2.2 DataFrame分组聚合后链式过滤的报错问题

嗨，我明白你遇到的问题了——直接用df.avg_x会报错，核心原因是**df还是你最初的原始DataFrame对象，它根本没有avg_x这个属性**。链式调用里每个步骤都会返回新的DataFrame，你得针对聚合后的新DataFrame来引用列，而不是揪着原始的df不放~

下面是两种不需要中间变量的链式写法，完美匹配你的需求：

方法1：用col()函数引用聚合列 #

先通过agg()给平均值列起别名avg_x，再用pyspark.sql.functions.col()来定位这个新列做过滤：

from pyspark.sql import functions as F

df.groupBy("country")
  .agg(F.avg("x").alias("avg_x"))
  .filter(F.col("avg_x") > 10)

方法2：用SQL风格字符串过滤 #

这种写法更简洁，直接在filter()里写类SQL的表达式就行：

from pyspark.sql import functions as F

df.groupBy("country")
  .agg(F.avg("x").alias("avg_x"))
  .filter("avg_x > 10")

补个小解释：为啥你的原写法不行？ #

你试的filter(df.avg_x > 10)里，df指向的是最开始的DataFrame（Schema只有country、year、x），它的列里压根没avg_x，自然会抛属性错误。而链式调用中agg()返回的是一个全新的DataFrame（包含country和avg_x），上面两种方法都是针对这个新对象的列操作，所以能正常运行。

内容的提问来源于stack exchange，提问作者M. Page