使用 Pyspark 更新 Dataframe 列中的 Minute 和 Seconds 值

Question

我的DF如下：

Name    starttime               endtime
user1   2019-08-02 03:34:45   2019-08-02 03:52:03
user2   2019-08-13 13:34:10   2019-08-13 14:02:10

我想检查endtime 是否会渗入下一小时，如果确实如此，则将其更新到当前小时的最后一分钟和第二秒，如下所示。

Name    starttime               endtime
user1   2019-08-02 03:34:45   2019-08-02 03:52:03
user2   2019-08-13 13:34:10   2019-08-13 13:59:59

我可以使用 UDF 进行如下检查和替换，但我不想使用它们。

def adjust_end_hour(date):
    return date.replace(second=59,minute=59)

adjust_end_hour_udf = udf(adjust_end_hour, TimestampType())

df = df.\
   filter(df.endtime > adjust_end_hour_udf(df.starttime)).\
   withColumn('enddtime', adjust_end_hour_udf(df.starttime))

如何在 pyspark 中不使用 UDF 来做到这一点？

Answer 1

另一种解决方案是将 starttime 截断为小时，然后使用 SQL 语法 INTERVAL 像这样添加 59 秒和 59 分钟：

adjust_expr = "date_trunc('hour', starttime) + INTERVAL 59 seconds + INTERVAL 59 minutes"

df.withColumn("endtime",
              when(col("endtime") > expr(adjust_expr),
                   expr(adjust_expr)
                  ).otherwise(col("endtime"))
              )\
  .show()

给予：

+-----+-------------------+-------------------+
| name|          starttime|            endtime|
+-----+-------------------+-------------------+
|user1|2019-08-02 03:34:45|2019-08-02 03:52:03|
|user2|2019-08-13 13:34:10|2019-08-13 13:59:59|
+-----+-------------------+-------------------+

Answer 2

假设您的 DataFrame 具有以下架构：

df.printSchema()
#root
# |-- Name: string (nullable = true)
# |-- starttime: timestamp (nullable = true)
# |-- endtime: timestamp (nullable = true)

即其中starttime 和endtime 都是TimestampType()。

您可以通过比较starttime 和endtime 的hour 部分来检查endtime 是否会持续到下一小时。如果它们不等于¹，则意味着您需要截断结束时间。

from pyspark.sql.functions import col, hour

df.withColumn(
    "bleeds_into_next_hour", 
    hour(col("endtime")) != hour(col("starttime"))
).show()
#+-----+-------------------+-------------------+---------------------+
#| Name|          starttime|            endtime|bleeds_into_next_hour|
#+-----+-------------------+-------------------+---------------------+
#|user1|2019-08-02 03:34:45|2019-08-02 03:52:03|                false|
#|user2|2019-08-13 13:34:10|2019-08-13 14:02:10|                 true|
#+-----+-------------------+-------------------+---------------------+

这会告诉您需要修改哪些行。您几乎可以使用date_trunc 将format 参数设置为hour 来获得所需的输出：

from pyspark.sql.functions import date_trunc, when

df.withColumn(
    "bleeds_into_next_hour", 
    hour(col("endtime")) != hour(col("starttime"))
).withColumn(
    "endtime", 
    when(
        col("bleeds_into_next_hour"), 
        date_trunc('hour', "endtime")
    ).otherwise(col("endtime"))
).show()
#+-----+-------------------+-------------------+---------------------+
#| Name|          starttime|            endtime|bleeds_into_next_hour|
#+-----+-------------------+-------------------+---------------------+
#|user1|2019-08-02 03:34:45|2019-08-02 03:52:03|                false|
#|user2|2019-08-13 13:34:10|2019-08-13 14:00:00|                 true|
#+-----+-------------------+-------------------+---------------------+

您现在所要做的就是从endtime 中减去 1 秒。最简单的方法是转换 unix_timestamp，减去 1，然后使用 from_unixtime 转换回来。

from pyspark.sql.functions import from_unixtime, unix_timestamp

df.withColumn(
    "bleeds_into_next_hour", 
    hour(col("endtime")) != hour(col("starttime"))
).withColumn(
    "endtime", 
    from_unixtime(
        unix_timestamp(
            when(
                col("bleeds_into_next_hour"), 
                date_trunc('hour', "endtime")
            ).otherwise(col("endtime"))
        ) - 1
    )
).drop("bleeds_into_next_hour").show()
#+-----+-------------------+-------------------+
#| Name|          starttime|            endtime|
#+-----+-------------------+-------------------+
#|user1|2019-08-02 03:34:45|2019-08-02 03:52:02|
#|user2|2019-08-13 13:34:10|2019-08-13 13:59:59|
#+-----+-------------------+-------------------+

---

把它们放在一起，没有中间列：

from pyspark.sql.functions import col, date_trunc, from_unixtime, hour, unix_timestamp, when

df = df.withColumn(
    "endtime", 
    from_unixtime(
        unix_timestamp(
            when(
                hour(col("endtime")) != hour(col("starttime")), 
                date_trunc('hour', "endtime")
            ).otherwise(col("endtime"))
        ) - 1
    )
)

备注

1. 假设endtime 总是大于或等于starttime。你不能这样做>，因为时间在 12 小时后结束。