在不減少行數的情況下在火花上執行max（）操作

Question

假設我們有一個包含三列（客戶ID，操作和操作時間）的數據集。

1, ACTION_1, 100 
1, ACTION_2, 101 
1, ACTION_3, 102 
2, ACTION_1, 100 
2, ACTION_2, 105 
2, ACTION_3, 102 
3, ACTION_1, 120 
3, ACTION_2, 111 
3, ACTION_3, 103 

我們要同時過濾像ACTION_2一些具體的行動來獲得每個客戶的最後一個動作的時間。如下所示：

1, ACTION_2, 102 
2, ACTION_2, 105 
3, ACTION_2, 120 

我們期待了解此問題的任何類型的解決方案。

Answer 1

創建數據框：

from pyspark.sql import HiveContext 
sqlContext = HiveContext(sc) 

data = [ 
(1, 'ACTION_1', 100), 
(1, 'ACTION_2', 101), 
(1, 'ACTION_3', 102), 
(2, 'ACTION_1', 100), 
(2, 'ACTION_2', 105), 
(2, 'ACTION_3', 102), 
(3, 'ACTION_1', 120), 
(3, 'ACTION_2', 111), 
(3, 'ACTION_3', 103)] 

df = sqlContext.createDataFrame(data, ['customerid', 'action', 'actiontime']) 
df.show() 

使用過窗口max函數由客戶ID

from pyspark.sql import Window 
from pyspark.sql.functions import max 
w = Window.partitionBy(df.customerid) 

df1 = df.withColumn('actiontime', max('actiontime').over(w)) 
df1.show() 

過濾的條件的數據進行分配：

df2 = df1.where(df1.action == 'ACTION_2') 
df2.show() 
+----------+--------+----------+ 
|customerid| action|actiontime| 
+----------+--------+----------+ 
|   1|ACTION_2|  102| 
|   3|ACTION_2|  120| 
|   2|ACTION_2|  105| 
+----------+--------+----------+