Pyspark根據row_number值增加時間戳列

我從 eventthub 中提取資料，每個資料包中有 10 條記錄，每個資料包上都有一個時間戳。我想分解由 10 條記錄組成的資料包，并且我想在按 EnqueuedTimeUtc 和 vehicleid 磁區時將資料包時間戳添加到每條記錄中，增量為 1 秒

下面是我在資料框中的中間資料。

df.show()

 ------------------- --------------- ------------------- 
|    EnqueuedTimeUtc|      vehicleid|   datetime_pkt    |
 ------------------- --------------- ------------------- 
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|
 ------------------- --------------- ------------------- 

預期產出

 ------------------- --------------- ------------------- ------------------- 
|    EnqueuedTimeUtc|      vehicleid|   datetime_pkt    | nw_datetime_pkt   |
 ------------------- --------------- ------------------- ------------------- 
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:19:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:20:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:21:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:22:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:23:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:24:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:25:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:26:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:27:43|
|5/1/2022 7:19:46 AM|86135903910    |2022-05-01 07:19:43|2022-05-01 07:28:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:19:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:20:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:21:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:22:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:23:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:24:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:25:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:26:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:27:43|
|5/1/2022 7:19:49 AM|86135903910    |2022-05-01 07:19:48|2022-05-01 07:28:43|
 ------------------- --------------- ------------------- ------------------- 

uj5u.com熱心網友回復：

我能夠通過使用視窗函式來解決上述任務。

腳步：

1. 為 partitionBy 列添加 row_number 并減去 1，以便 row_number 從 0 而不是 1 開始。
2. 利用滯后函式并創建一個新列 nw_datetime_pkt。
3. 利用 unix_timestamp 函式，該函式采用時間戳列和秒數遞增

import pyspark.sql.functions as F

df1 = df.withColumn("rn", F.row_number().over(Window.partitionBy("vehicleid", "datetime_pkt").orderBy("datetime_pkt")) - 1) \
        .withColumn("nw_datetime_pkt", F.lag(F.col("datetime_pkt")).over(Window.partitionBy("vehicleid", "datetime_pkt").orderBy("datetime_pkt")))

df1 = df1.withColumn("nw_datetime_pkt", F.when(F.col("nw_datetime_pkt").isNull(), F.col("datetime_pkt")).otherwise((F.unix_timestamp("nw_datetime_pkt")   df1.rn).cast('timestamp')))

標籤：阿帕奇火花 pyspark 数据块

上一篇：使用類似“all”的函式pyspark聚合GroupBy列

下一篇：如何在承諾中呼叫fetch？