将时间戳舍入到最接近的 30 秒

Question

我在 DF 中有一列，它包含格式为 (yyyy-mm-dd HH:mm:ss) 的 timestamp。我需要将 timestamp 舍入到最接近的 30 秒。

old column                   desired column
2016-02-09 19:31:02          2016-02-09 19:31:00  
2016-02-09 19:31:35          2016-02-09 19:31:30
2016-02-09 19:31:52          2016-02-09 19:32:00
2016-02-09 19:31:28          2016-02-09 19:31:30

在 Pyspark 中可以这样做吗？

Answer 1

如果您使用的是 spark verson 1.5+，您可以使用 pyspark.sql.functions.second() 从时间戳列中获取秒数。

import pyspark.sql.functions as f
df.withColumn("second", f.second("old_timestamp")).show()
#+-------------------+------+
#|      old_timestamp|second|
#+-------------------+------+
#|2016-02-09 19:31:02|     2|
#|2016-02-09 19:31:35|    35|
#|2016-02-09 19:31:52|    52|
#|2016-02-09 19:31:28|    28|
#+-------------------+------+

一旦你有了秒部分，你就可以取这个数字，除以 30，四舍五入，然后乘以 30 得到 "new" 秒。

df.withColumn("second", f.second("old_timestamp"))\
    .withColumn("new_second", f.round(f.col("second")/30)*30)\
    .show()
#+-------------------+------+----------+
#|      old_timestamp|second|new_second|
#+-------------------+------+----------+
#|2016-02-09 19:31:02|     2|       0.0|
#|2016-02-09 19:31:35|    35|      30.0|
#|2016-02-09 19:31:52|    52|      60.0|
#|2016-02-09 19:31:28|    28|      30.0|
#+-------------------+------+----------+

从 "new" 秒开始，我们可以计算出以秒为单位的偏移量，将其添加到原始时间戳后将产生所需的 "rounded" 时间戳。

df.withColumn("second", f.second("old_timestamp"))\
    .withColumn("new_second", f.round(f.col("second")/30)*30)\
    .withColumn("add_seconds", f.col("new_second") - f.col("second"))\
    .show()
#+-------------------+------+----------+-----------+
#|      old_timestamp|second|new_second|add_seconds|
#+-------------------+------+----------+-----------+
#|2016-02-09 19:31:02|     2|       0.0|       -2.0|
#|2016-02-09 19:31:35|    35|      30.0|       -5.0|
#|2016-02-09 19:31:52|    52|      60.0|        8.0|
#|2016-02-09 19:31:28|    28|      30.0|        2.0|
#+-------------------+------+----------+-----------+

如我们所见，此列中的负数表示原始时间必须向下舍入。正数将增加时间。

为了将这个时间添加到原始时间戳中，首先使用 pyspark.sql.functions.unix_timestamp(). After the addition, convert the result back to a timestamp using pyspark.sql.functions.from_unixtime() 将其转换为 unix 时间戳。

将这些放在一起（压缩中间步骤）：

df.withColumn(
        "add_seconds",
        (f.round(f.second("old_timestamp")/30)*30) - f.second("old_timestamp")
    )\
    .withColumn(
        "new_timestamp",
        f.from_unixtime(f.unix_timestamp("old_timestamp") + f.col("add_seconds"))
    )\
    .drop("add_seconds")\
    .show()
#+-------------------+-------------------+
#|      old_timestamp|      new_timestamp|
#+-------------------+-------------------+
#|2016-02-09 19:31:02|2016-02-09 19:31:00|
#|2016-02-09 19:31:35|2016-02-09 19:31:30|
#|2016-02-09 19:31:52|2016-02-09 19:32:00|
#|2016-02-09 19:31:28|2016-02-09 19:31:30|
#+-------------------+-------------------+

将时间戳舍入到最接近的 30 秒

Round timestamp to nearest 30 second

python

timestamp

unix-timestamp

pyspark