在不减少行数的情况下对spark执行max（）操作

Question

假设我们有一个由三列组成的数据集，分别是customer id、action和action time。你知道吗

 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

我们想得到每个客户的最后一个动作时间，同时过滤一些特定的动作，比如动作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()

在按客户id划分的窗口上使用max函数

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|
+     +    +     +