如何使用DataFrame在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| 
## +---+---+-----+ 

一般來說星火SQL窗口函數的行爲與任何現代RDBMS中的方式完全相同。