python - 如何使用 DataFrames 在 PySpark 中使用窗口函数？

Question

试图找出如何在 PySpark 中使用窗口函数。这是我希望能够执行的操作的示例，只需计算用户发生“事件”的次数(在本例中“dt”是模拟时间戳)。

from pyspark.sql.window import Window
from pyspark.sql.functions import count

df = sqlContext.createDataFrame([{"id": 123, "dt": 0}, {"id": 123, "dt": 1}, {"id": 234, "dt":0}, {"id": 456, "dt":0}, {"id": 456, "dt":1}, {"id":456, "dt":2}])
df.select(["id","dt"], count("dt").over(Window.partitionBy("id").orderBy("dt")).alias("count")).show()

这会产生一个错误。窗口函数的正确使用方法是什么？我读到 1.4.1(我们需要使用的版本，因为它是 AWS 上的标准版本)应该能够使用 DataFrame API 来完成它们。

FWIW，关于这个主题的文档非常少。而且我很难让任何示例实际运行。

Answer 1

它会抛出异常，因为您传递了一个列列表。 DataFrame.select 的签名如下所示

df.select(self, *cols)

并且使用窗口函数的表达式是一个与其他任何列一样的列，因此您在这里需要的是这样的:

w = Window.partitionBy("id").orderBy("dt") # Just for clarity
df.select("id","dt", count("dt").over(w).alias("count")).show()

## +---+---+-----+
## | id| dt|count|
## +---+---+-----+
## |234|  0|    1|
## |456|  0|    1|
## |456|  1|    2|
## |456|  2|    3|
## |123|  0|    1|
## |123|  1|    2|
## +---+---+-----+

一般来说，Spark SQL 窗口函数的行为方式与任何现代 RDBMS 中的行为方式完全相同。