按 24 小时划分并使用 pyspark 或 panda 聚合

Posted 2023-04-15

【中文标题】按 24 小时划分并使用 pyspark 或 panda 聚合

【英文标题】：partition by 24 hours and aggregate using pyspark or panda

【发布时间】：2020-04-21 14:52:30

【问题描述】：

我有每个设备的会话数据，如下所示

time-started：设备连接时的时间戳

持续时间（秒）：连接的时间

packets：发送了多少包

现在我需要按 24 小时汇总（总和）每个设备分区的持续时间和数据包。

例如：

对于第一个记录，

设备 A，时间从 4 月 8 日凌晨 1:53 开始，因此需要汇总所有设备 (A)，有效期为 24 小时，即 4 月 9 日凌晨 1:53。

那么设备 A 的下一条记录应该从 4 月 9 日凌晨 4 点 27 分开始，因此需要汇总所有 24 小时有效的设备 (A)，直到 4 月 10 日凌晨 4 点 27 分。

和

每个设备都如此。

预期输出

测试数据：

dftest = sc.parallelize([['A','2020-04-08T01:53:54.932000','Org1','wifi',60,372717],
                      ['A','2020-04-08T02:40:38.661000','Org1','wifi',194,819040],
                       ['A','2020-04-08T21:45:10.207000','Org1','wifi',8885,3449150],
                        ['A','2020-04-09T00:15:28.838000','Org1','wifi',14770,3572589],
                         ['A','2020-04-09T04:27:33.424000','Org1','remote',0,0],
                          ['A','2020-04-09T04:29:25.189000','Org1','wifi',60,7495],
                           ['A','2020-04-09T04:44:21.397000','Org1','remote',60,553356],
                            ['A','2020-04-09T04:50:40.406000','Org1','wifi',60,662467],
                             ['A','2020-04-10T00:00:50.636000','Org1','remote',0,72],
                              ['A','2020-04-10T04:20:28.831000','Org1','remote',6,497],
                               ['A','2020-04-10T04:31:35.336000','Org1','remote',0,22],
                                ['B','2020-04-08T21:56:58.502000','Org2','remote',0,0],
                                 ['B','2020-04-08T22:01:19.534000','Org2','wifi',0,0],
                                  ['B','2020-04-08T22:10:15.891000','Org2','wifi',60,187891],
                                   ['B','2020-04-08T22:16:41.499000','Org2','wifi',1620,207674],
                                    ['B','2020-04-09T01:55:02.772000','Org2','wifi',360,426232],
                                     ['B','2020-04-09T02:03:32.735000','Org2','wifi',60,374827],
                                      ['B','2020-04-09T02:06:16.509000','Org2','wifi',60,386518],
                                       ['B','2020-04-09T02:13:33.497000','Org2','remote',60,373609],
                                        ['B','2020-04-09T02:17:19.176000','Org2','wifi',133,400417],
                                         ['B','2020-04-10T23:10:15.654000','Org2','remote',0,212],
                                          ['B','2020-04-10T23:10:41.749000','Org2','remote',1,285]
                    ]).toDF(("deviceId","time-started","OrgId","type","duration","packet"))
dftest.show()

【问题讨论】：

partition device by 24 hours 意味着第一个 A 从 8thApril 1:53 AM 开始，第二个在 2020-04-09T04:27 ，3rd 在 2020-04-10T04:20 而不是 4:31 因为第一次出现来自8th 1:53-9th 1:53，第二个来自下一个值，即9th 4:27 untill 10th 1:53，第三个来自下一个值，即10th 4:27,,,.....否？

@raviteja 我认为你的 24 小时窗口对于 spark 窗口函数来说有点动态，因为它取决于最后一个的结束来计算下一个，建议使用 spark 窗口函数和组合熊猫 udaf。

@MohammadMurtazaHashmi 你能详细说明一下

@RaviTeja 我刚刚发布，如果有帮助，请 lmk

【参考方案1】：

对于您的情况，您的下一个 24 hour 取决于 end of the last one 和 date after that last date，因此我们不能只用窗口函数来表达这个逻辑。我decoupled从pandas计算总和（因为它会很慢）并使用spark in-built functions to get your sum，pandas udaf基本上给了我们我们的desired date groups，我们对它们filter 以获得所需的结果。

迭代是获得24小时段的唯一方法，因此您也可以使用简单的 udf，但是 pandas 矢量化 udaf 允许我们在小组中表达我们的逻辑（基于id) 因此应该更好。 Pandas-udaf(spark2.3+)

import pandas as pd
from pyspark.sql.functions import pandas_udf, PandasUDFType
from pyspark.sql import functions as F
from pyspark.sql.window import Window

w=Window().partitionBy("deviceId").orderBy(F.col("time-started").cast("long")).rangeBetween(Window.currentRow,24*60*60)
df2=df.withColumn("time-started", F.to_timestamp("time-started", "yyyy-MM-dd'T'HH:mm:ss"))\
      .withColumn("time-started-2", F.col("time-started"))\
      .withColumn("duration", F.sum("duration").over(w))\
      .withColumn("packet", F.sum("packet").over(w))

@pandas_udf(df2.schema, PandasUDFType.GROUPED_MAP)
def grouped_map(df1):
   start=df1.loc[0, 'time-started']
   for i in range(1, len(df1)):
        if start + pd.Timedelta(days=1)>df1.loc[i,'time-started']:
             df1.loc[i,'time-started']=start
        else:
             start=df1.loc[i,'time-started']    


   return df1
df2.groupby('deviceId').apply(grouped_map)\
.filter(F.col("time-started-2")==F.col("time-started"))\
.drop("time-started-2")\
.orderBy("deviceId")\
.show()

#+--------+-------------------+-----+------+--------+-------+
#|deviceId|       time-started|OrgId|  type|duration| packet|
#+--------+-------------------+-----+------+--------+-------+
#|       A|2020-04-08 01:53:54| Org1|  wifi|   23909|8213496|
#|       A|2020-04-09 04:27:33| Org1|remote|     186|1223887|
#|       A|2020-04-10 04:31:35| Org1|remote|       0|     22|
#|       B|2020-04-08 21:56:58| Org2|remote|    2353|2357168|
#|       B|2020-04-10 23:10:15| Org2|remote|       1|    497|
#+--------+-------------------+-----+------+--------+-------+

您也可以查看similar question。建议的解决方案是使用 foldleft 功能的 scala udf。我认为 pandas 分组地图 udaf 会是更好的选择。