如何根据pysp中的前一行值向给定的时间戳添加分钟

Question

我有一个pyspark数据帧

   +----------+----------+---------------------+
   | Activity | Interval |    ReadDateTime     |
   +----------+----------+---------------------+
   |    A     |    1     | 2019-12-13 10:00:00 | 
   |    A     |    2     | 2019-12-13 10:00:00 |
   |    A     |    3     | 2019-12-13 10:00:00 |
   |    B     |    1     | 2019-12-13 11:00:00 | 
   |    B     |    2     | 2019-12-13 11:00:00 |
   |    B     |    3     | 2019-12-13 11:00:00 |
   +--------- +----------+---------------------+

现在，我必须根据上一行中的值向ReadDateTime列添加5分钟。我期望的数据帧如下所示

^{pr2}$

我不会将5分钟添加到与间隔1对应的ReadDateTime列中，而将继续向其他行添加5分钟，直到我的活动发生更改

Answer 1

有一个丑陋的方法

from pyspark.sql.functions import *
from pyspark.sql.types import StringType

def update(interval,date):
  if (interval == 1):
    return date
  elif (interval == 2):
    return date + 'add 5 min'
  elif (interval == 3):
    return date + 'add 10 min'

#df.dtypes

my_udf = udf(lambda x,y: update(x,y), StringType())

df.withColumn('updated_realDateTime', my_udf(df.interval, df.realDateTime) ).show(truncate=False)

当然，我的更新功能不是你想要的，所以你必须改变它，但它会完成工作（你不需要elifs如果模式在所有间隔都是相同的，你可以使它动态）

下面是为任何有更好答案的人创建数据帧的代码

^{pr2}$

Answer 2

多亏了阿里·耶斯利的帖子，我已经想出了解决办法 adding hours to timestamp in pyspark dymanically。 我首先将ReadDateTime转换为unix时间戳，并且只有在间隔不等于1时才添加5分钟。所以我的代码如下所示。在

   from pyspark.sql.functions import col,explode,lit
   from pyspark.sql import functions as F

   df = df.withColumn("ReadDateTime1", F.when(col("Interval") != lit(1)
   F.unix_timestamp("ReadDateTime") + 
   (F.col("a")-1)*300).cast('timestamp')).otherwise(col('ReadDateTime')))