pyspark：計算不同的窗口上面

Question

我只是想在窗口做一個countDistinct和得到這個錯誤：

AnalysisException: u'Distinct window functions are not supported: count(distinct color#1926) 

有沒有辦法在在pyspark窗口做一個重複計數？

下面是一些示例代碼：

from pyspark.sql import functions as F 

#function to calculate number of seconds from number of days 
days = lambda i: i * 86400 

df = spark.createDataFrame([(17, "2017-03-10T15:27:18+00:00", "orange"), 
        (13, "2017-03-15T12:27:18+00:00", "red"), 
        (25, "2017-03-18T11:27:18+00:00", "red")], 
        ["dollars", "timestampGMT", "color"]) 

df = df.withColumn('timestampGMT', df.timestampGMT.cast('timestamp')) 

#create window by casting timestamp to long (number of seconds) 
w = (Window.orderBy(F.col("timestampGMT").cast('long')).rangeBetween(-days(7), 0)) 

df = df.withColumn('distinct_color_count_over_the_last_week', F.countDistinct("color").over(w)) 

df.show() 

這是我希望看到的輸出：

+-------+--------------------+------+---------------------------------------+ 
|dollars|  timestampGMT| color|distinct_color_count_over_the_last_week| 
+-------+--------------------+------+---------------------------------------+ 
|  17|2017-03-10 15:27:...|orange|          1| 
|  13|2017-03-15 12:27:...| red|          2| 
|  25|2017-03-18 11:27:...| red|          1| 
+-------+--------------------+------+---------------------------------------+ 

Answer 1

我想通了，我可以使用的collect_set和大小功能的組合在窗口中模擬countDistinct的功能：

from pyspark.sql import functions as F 

#function to calculate number of seconds from number of days 
days = lambda i: i * 86400 

#create some test data 
df = spark.createDataFrame([(17, "2017-03-10T15:27:18+00:00", "orange"), 
        (13, "2017-03-15T12:27:18+00:00", "red"), 
        (25, "2017-03-18T11:27:18+00:00", "red")], 
        ["dollars", "timestampGMT", "color"]) 

#convert string timestamp to timestamp type    
df = df.withColumn('timestampGMT', df.timestampGMT.cast('timestamp')) 

#create window by casting timestamp to long (number of seconds) 
w = (Window.orderBy(F.col("timestampGMT").cast('long')).rangeBetween(-days(7), 0)) 

#use collect_set and size functions to perform countDistinct over a window 
df = df.withColumn('distinct_color_count_over_the_last_week', F.size(F.collect_set("color").over(w))) 

df.show() 

這會導致不同的顏色計數o ver前一週的記錄：

+-------+--------------------+------+---------------------------------------+ 
|dollars|  timestampGMT| color|distinct_color_count_over_the_last_week| 
+-------+--------------------+------+---------------------------------------+ 
|  17|2017-03-10 15:27:...|orange|          1| 
|  13|2017-03-15 12:27:...| red|          2| 
|  25|2017-03-18 11:27:...| red|          1| 
+-------+--------------------+------+---------------------------------------+