如何按小時獲取匯總，包括丟失的小時數并添加累積總和？

假設我在下面有一個 Spark Dataframe：

我想根據GroupId在一天中每小時獲取 eventA 值的滾動計數

例如，對于GroupId xx，datetime 2011-08-15 14:00，嘗試計算從 14:00 到 15:00的 eventA ( Event_value )的計數。GroupId在這種情況下，計數應該是 1 2 = 3。

預期的輸出將類似于：（基本上在一天內顯示從 00 到 23；出于節省空間的目的，我忽略了以下幾個小時）。

如果沒有 eventA，那么我們將該小時范圍內的計數視為 NA（為稍后計算目的而視為 0）。

對于 event_date 2011-08-15，在 14 小時之前沒有事件，然后在 16 小時之后沒有更多事件。

以下是我嘗試過的一些代碼：

from pyspark.sql.functions import col, count, hour, sum
    
df2 = (df
  .withColumn("Event_time", col("Event_time").cast("timestamp"))
  .withColumn("Date", col("Event_time").cast("date"))
  .withColumn("Hour", hour(col("Event_time"))))

df3 = df2.groupBy("GroupId", "Date", "Hour").count()

df3.withColumn(
  "agg_count", 
  sum("Count").over(Window.partitionBy("GroupId", "Date").orderBy("Hour")))

但是，上面的代碼不能顯示一天內的每個小時。

uj5u.com熱心網友回復：

您可以先創建一個包含小時數的表，然后將其與其余資料連接起來。

設定：

from pyspark.sql import functions as F, Window as W
df = spark.createDataFrame(
    [('xx', '2011-08-15 14:47:02.617023', 'eventA', 1),
     ('xx', '2011-08-15 14:48:02.507053', 'eventA', 2),
     ('xx', '2011-08-15 16:47:02.512016', 'eventA', 100),
     ('yy', '2011-08-15 11:47:02.337019', 'eventA', 2),
     ('yy', '2011-08-15 12:47:02.617041', 'eventA', 1),
     ('yy', '2011-08-15 13:47:02.927040', 'eventA', 3)],
    ['GroupId', 'Event_time', 'Event_name', 'Event_value']
)
df = df.withColumn('Date', F.col('Event_time').cast('date'))

以下創建了一個帶有小時的資料框：

min_date = df.groupBy().agg(F.min('Date')).head()[0]
max_date = df.groupBy().agg(F.max('Date')).head()[0]
df_hours = df.select(
    'GroupId',
    'Event_name',
    F.explode(F.expr(f"sequence(to_timestamp('{min_date} 00:00:00'), to_timestamp('{max_date} 23:00:00'), interval 1 hour)")).alias('date_hour')
).distinct()

然后，按小時聚合您的第一個表：

df_agg = (df
    .groupBy('GroupId', 'Event_name', F.date_trunc('hour', 'Event_time').alias('date_hour'))
    .agg(F.sum('Event_value').alias('Count'))
)

將它們結合在一起：

df_joined = df_hours.join(df_agg, ['GroupId', 'Event_name', 'date_hour'], 'left')

添加列agg_count和其他：

w = W.partitionBy('GroupId', 'Event_name').orderBy('date_hour')
df2 = (df_joined
    .select(
        'GroupId',
        'Event_name',
        F.to_date('date_hour').alias('Date'),
        F.date_format('date_hour', 'HH').alias('Hour'),
        'Count',
        F.coalesce(F.sum('Count').over(w), F.lit(0)).alias('agg_count')
    )
)

結果：

 ------- ---------- ---------- ---- ----- --------- 
|GroupId|Event_name|      Date|Hour|Count|agg_count|
 ------- ---------- ---------- ---- ----- --------- 
|     xx|    eventA|2011-08-15|  00| null|        0|
|     xx|    eventA|2011-08-15|  01| null|        0|
|     xx|    eventA|2011-08-15|  02| null|        0|
|     xx|    eventA|2011-08-15|  03| null|        0|
|     xx|    eventA|2011-08-15|  04| null|        0|
|     xx|    eventA|2011-08-15|  05| null|        0|
|     xx|    eventA|2011-08-15|  06| null|        0|
|     xx|    eventA|2011-08-15|  07| null|        0|
|     xx|    eventA|2011-08-15|  08| null|        0|
|     xx|    eventA|2011-08-15|  09| null|        0|
|     xx|    eventA|2011-08-15|  10| null|        0|
|     xx|    eventA|2011-08-15|  11| null|        0|
|     xx|    eventA|2011-08-15|  12| null|        0|
|     xx|    eventA|2011-08-15|  13| null|        0|
|     xx|    eventA|2011-08-15|  14|    3|        3|
|     xx|    eventA|2011-08-15|  15| null|        3|
|     xx|    eventA|2011-08-15|  16|  100|      103|
|     xx|    eventA|2011-08-15|  17| null|      103|
|     xx|    eventA|2011-08-15|  18| null|      103|
|     xx|    eventA|2011-08-15|  19| null|      103|
|     xx|    eventA|2011-08-15|  20| null|      103|
|     xx|    eventA|2011-08-15|  21| null|      103|
|     xx|    eventA|2011-08-15|  22| null|      103|
|     xx|    eventA|2011-08-15|  23| null|      103|
|     yy|    eventA|2011-08-15|  00| null|        0|
|     yy|    eventA|2011-08-15|  01| null|        0|
|     yy|    eventA|2011-08-15|  02| null|        0|
|     yy|    eventA|2011-08-15|  03| null|        0|
|     yy|    eventA|2011-08-15|  04| null|        0|
|     yy|    eventA|2011-08-15|  05| null|        0|
|     yy|    eventA|2011-08-15|  06| null|        0|
|     yy|    eventA|2011-08-15|  07| null|        0|
|     yy|    eventA|2011-08-15|  08| null|        0|
|     yy|    eventA|2011-08-15|  09| null|        0|
|     yy|    eventA|2011-08-15|  10| null|        0|
|     yy|    eventA|2011-08-15|  11|    2|        2|
|     yy|    eventA|2011-08-15|  12|    1|        3|
|     yy|    eventA|2011-08-15|  13|    3|        6|
|     yy|    eventA|2011-08-15|  14| null|        6|
|     yy|    eventA|2011-08-15|  15| null|        6|
|     yy|    eventA|2011-08-15|  16| null|        6|
|     yy|    eventA|2011-08-15|  17| null|        6|
|     yy|    eventA|2011-08-15|  18| null|        6|
|     yy|    eventA|2011-08-15|  19| null|        6|
|     yy|    eventA|2011-08-15|  20| null|        6|
|     yy|    eventA|2011-08-15|  21| null|        6|
|     yy|    eventA|2011-08-15|  22| null|        6|
|     yy|    eventA|2011-08-15|  23| null|        6|
 ------- ---------- ---------- ---- ----- --------- 

標籤：阿帕奇火花 pyspark apache-spark-sql 总计的 窗函数

上一篇：用初始化代碼撰寫pysparkUDF

下一篇：在intellij上創建一個scala專案