我有以下 Spark 數據框： id | time |Value| id1 | 2020-02-22 04:57:36.843 | 1.4 | id2 | 2020-02-22 04:57:50.850 | 1.7 | id3 | 2020-02-22 04:58:02.133 | 1.2 |我想在現有行之間插入一定的及時距離（例如5秒）。輸出應如下所示： id | time |Value| id1 | 2020-02-22 04:57:36.843 | 1.4 | id1 | 2020-02-22 04:57:41.843 | | id1 | 2020-02-22 04:57:46.843 | | id1 | 2020-02-22 04:57:51.843 | | id2 | 2020-02-22 04:57:50.850 | 1.7 | id2 | 2020-02-22 04:57:55.850 | | id2 | 2020-02-22 04:58:00.850 | | id2 | 2020-02-22 04:58:05.850 | | id3 | 2020-02-22 04:58:02.133 | 1.2 | id3 | 2020-02-22 04:58:07.133 | | id3 | 2020-02-22 04:58:12.133 | | id3 | 2020-02-22 04:58:17.133 | |我嘗試通過 for 循環來實現這一點，創建新的數據幀（每個新行）并通過“union”將其合并到現有的數據幀，但沒有成功。我尤其沒有通過這種方法獲得 id。你知道我如何達到我想要的輸出嗎？
1 回答
人到中年有點甜
TA貢獻1895條經驗 獲得超7個贊
這是我嘗試進行一些修改，例如，我無法理解如何存在 62 秒。
from pyspark.sql.functions import *
from pyspark.sql import Window
w = Window.orderBy('time')
df.select('id', 'time') \
.withColumn('time', to_timestamp('time', 'yyyy-MM-dd HH:mm:ss.SSS')) \
.withColumn('time2', coalesce(lead('time', 1).over(w), expr('time + interval 10 seconds'))) \
.withColumn('seq', expr("sequence(time, time2 + interval 5 seconds, interval 5 seconds)")) \
.withColumn('time', explode('seq')) \
.select('id', 'time') \
.join(df, ['id', 'time'], 'left') \
.fillna(0).show(20, False)
+---+-----------------------+-----+
|id |time |Value|
+---+-----------------------+-----+
|id1|2020-02-22 04:57:36.843|1.4 |
|id1|2020-02-22 04:57:41.843|0.0 |
|id1|2020-02-22 04:57:46.843|0.0 |
|id1|2020-02-22 04:57:51.843|0.0 |
|id2|2020-02-22 04:57:50.85 |1.7 |
|id2|2020-02-22 04:57:55.85 |0.0 |
|id2|2020-02-22 04:58:00.85 |0.0 |
|id3|2020-02-22 04:57:59.133|1.2 |
|id3|2020-02-22 04:58:04.133|0.0 |
|id3|2020-02-22 04:58:09.133|0.0 |
|id3|2020-02-22 04:58:14.133|0.0 |
+---+-----------------------+-----+
添加回答
舉報
0/150
提交
取消