Pyspark - 以整数表示的舍入时间到最接近的一刻钟（15 分钟）答案

【问题标题】：Pyspark - Round time represented as an integer to nearest quarter hour (15 minutes)Pyspark - 以整数表示的舍入时间到最接近的一刻钟（15 分钟）
【发布时间】：2018-05-15 08:07:55
【问题描述】：

我正在研究 Pyspark 上的数据框。一列由对应时间的整数值组成：

data.select('TIME').show(4)
+------------------+
|TIME              |
+------------------+
|               925|
|              2205|
|              2205|
|              2205|
+------------------+

我想以15分钟的时间步长将这个时间四舍五入到最接近的值，以获得：

+------------------+
|TIME_15_MIN_STEP  |
+------------------+
|               930|
|              2200|
|              2200|
|              2200|
+------------------+

有人知道怎么做吗？

非常感谢！！

【问题讨论】：

我猜对于 925，算法应该检查 925 是否更接近 930 或 915，这是 15 的最接近的倍数。我不明白为什么 2205 必须变为 2200？ 2205已经是15的倍数了，这个四舍五入的逻辑是什么？
因为它们代表时间（2205 = 22:05 = 10:05 pm）。
将时间表示为整数 (925) 而不是字符串 ("09:25") 是否有一些具体原因？如果时间是时间戳（例如，从纪元开始的毫秒数），则将时间表示为整数很方便；如果是小时或日期，则将时间表示为字符串（并且在这两种情况下都可以执行许多操作）
使用 UDF。第一步将这次转换为普通datetime，然后使用它stackoverflow.com/questions/32723150/…

标签： apache-spark pyspark apache-spark-sql rounding

【解决方案1】：

按 15 分钟分组的更好方法是在时间戳上使用 pyspark.sql.functions.window：

df = df \
   .groupBy(F.window("timestamp", "15 minutes")) \
   .withColumn("timestamp", F.col("window.start"))

参见文档here

【讨论】：

【解决方案2】：

不使用udf 的一种方法是首先将整数列转换为虚拟时间戳，然后执行与my answer 中概述的几乎相同的操作similar question。最后将结果转换回所需格式的整数。

更详尽的示例数据

我创建了一个具有更多可变性的示例，以证明此方法可以正常工作。

data = sqlCtx.createDataFrame([(925,), (2205,), (2210,), (2242,), (2255,)], ["TIME"])
data.show()
#+----+
#|TIME|
#+----+
#| 925|
#|2205|
#|2210|
#|2242|
#|2255|
#+----+

将整数列转换为虚拟时间戳

要将整数小时-分钟列转换为时间戳，我们首先使用pyspark.sql.functions.format_string() 到add leading zeros 到适当的时间。接下来我们将一个虚拟日期（我使用"2018-01-01"）与转换后的时间连接起来，并在末尾添加":00"（以秒为单位）。

data = data.withColumn("time_string", f.format_string("%04d", f.col("TIME")))\
    .withColumn(
        "time_string",
        f.concat_ws(
            ":",
            f.array(
                [
                    f.substring(
                        "time_string",
                        1,
                        2
                    ),
                    f.substring(
                        "time_string",
                        3,
                        2
                    ),
                    f.lit("00")
                ]
            )
        )
    )\
    .withColumn("time_string", f.concat(f.lit("2018-01-01 "), f.col("time_string")))

data.show()
#+----+-------------------+
#|TIME|        time_string|
#+----+-------------------+
#| 925|2018-01-01 09:25:00|
#|2205|2018-01-01 22:05:00|
#|2210|2018-01-01 22:10:00|
#|2242|2018-01-01 22:42:00|
#|2255|2018-01-01 22:55:00|
#+----+-------------------+

计算偏移时间戳的分钟数

使用pyspark.sql.functions.minute() 从虚拟时间戳中获取分钟。我们除以 15，四舍五入，然后乘以 15 得到“新”分钟。（这个逻辑在linked answer中有更详细的解释。）

data = data.withColumn("minute", f.minute("time_string"))\
    .withColumn("new_minute", f.round(f.col("minute")/15)*15)\
    .withColumn("minute_add", f.col("new_minute") - f.col("minute"))\

data.show()
#+----+-------------------+------+----------+----------+
#|TIME|        time_string|minute|new_minute|minute_add|
#+----+-------------------+------+----------+----------+
#| 925|2018-01-01 09:25:00|    25|      30.0|       5.0|
#|2205|2018-01-01 22:05:00|     5|       0.0|      -5.0|
#|2210|2018-01-01 22:10:00|    10|      15.0|       5.0|
#|2242|2018-01-01 22:42:00|    42|      45.0|       3.0|
#|2255|2018-01-01 22:55:00|    55|      60.0|       5.0|
#+----+-------------------+------+----------+----------+

以秒为单位添加偏移量，转换回整数

将minute_add 列乘以 60 以获得以秒为单位的偏移量。将此添加到time_string 以获得“新”时间。

data = data.withColumn(
        "new_time",
        f.from_unixtime(f.unix_timestamp("time_string") + f.col("minute_add")*60)
    )\
    .withColumn(
        "NEW_TIME",
        f.format_string("%02d%02d", f.hour("new_time"), f.minute("new_time")).cast("int")
    )
data.select("TIME", "NEW_TIME").show()
#+----+--------+
#|TIME|NEW_TIME|
#+----+--------+
#| 925|     930|
#|2205|    2200|
#|2210|    2215|
#|2242|    2245|
#|2255|    2300|
#+----+--------+

【讨论】：

你是一个学者，一个绅士。这是一个很好的答案，虽然它下面的那个看起来也不错。