我有一个带有日期时间列的 Spark 数据框，格式为 yyyy-MM-dd HH:mm:SS。
我的数据框如下所示：
| datetime_column | output_column |
|---|---|
| 2021-01-29 21:30:00 | 2021-01-30 00:00:00 |
| 2021-01-30 05:30:00 | 2021-01-30 00:00:00 |
| 2021-01-01 23:25:00 | 2021-01-02 00:00:00 |
| 2021-01-02 08:59:59 | 2021-01-02 00:00:00 |
如何使用 Spark SQL 从 datetime_column 列中获取 output_column 列？
注意：我的数据 (datetime_column) 将始终在晚上 9 点到早上 9 点之间出现，因此查找最近的午夜很简单。
我正在使用 PySpark 2.4.6
【问题讨论】：
标签： apache-spark pyspark apache-spark-sql
这是一个通过计算秒值中最接近的日期来执行“四舍五入”的技巧（即，round((timestamp // 86400) * 86400)。将数字四舍五入到最接近的 1000 与我所做的相同：round(number//1000)*1000）
(df.select(f.from_unixtime(f.round(
f.unix_timestamp('datetime_column') /
f.lit(86400)) * f.lit(86400)).alias('output_column') )).show()
输出：
+-------------------+
| output_column|
+-------------------+
|2021-01-30 02:00:00|
|2021-01-30 02:00:00|
|2021-01-02 02:00:00|
|2021-01-02 02:00:00|
+-------------------+
请注意，上面的结果显示在我的时区中，其偏移量为 +02:00（from_unixtime 转换为当前时区的时间戳）。如果您需要准确的结果，请务必在运行代码之前在 Spark 配置中将 UTC 设置为系统时区（这是 mentioned in the function's docs）：
spark.conf.set("spark.sql.session.timeZone", "UTC")
(df.select(f.from_unixtime(f.round(
f.unix_timestamp('datetime_column') /
f.lit(86400)) * f.lit(86400)).alias('output_column'))).show()
spark.conf.unset("spark.sql.session.timeZone") # unset it
这会产生您的预期输出
+-------------------+
| output_column|
+-------------------+
|2021-01-30 00:00:00|
|2021-01-30 00:00:00|
|2021-01-02 00:00:00|
|2021-01-02 00:00:00|
+-------------------+
【讨论】：
执行此操作的更直观方法 - 您可以使用 case when 检查小时，如果小时晚于晚上 9 点，则将日期加 1。否则保留日期并删除时间。
import pyspark.sql.functions as F
df2 = df.withColumn(
'output_column',
F.when(
F.hour('datetime_column') >= 21,
F.date_add(F.date_trunc('day', 'datetime_column'), 1)
).otherwise(F.date_trunc('day', 'datetime_column'))
)
df2.show()
+-------------------+-------------------+
| datetime_column| output_column|
+-------------------+-------------------+
|2021-01-29 21:30:00|2021-01-30 00:00:00|
|2021-01-30 05:30:00|2021-01-30 00:00:00|
|2021-01-01 23:25:00|2021-01-02 00:00:00|
|2021-01-02 08:59:59|2021-01-02 00:00:00|
+-------------------+-------------------+
如果您更喜欢 Spark SQL：
df.createOrReplaceTempView('df')
df2 = spark.sql("""
select *,
case when hour(datetime_column) >= 21
then date_add(date_trunc('day', datetime_column), 1)
else date_trunc('day', datetime_column)
end as output_column
from df
""")
df2.show()
+-------------------+-------------------+
| datetime_column| output_column|
+-------------------+-------------------+
|2021-01-29 21:30:00|2021-01-30 00:00:00|
|2021-01-30 05:30:00|2021-01-30 00:00:00|
|2021-01-01 23:25:00|2021-01-02 00:00:00|
|2021-01-02 08:59:59|2021-01-02 00:00:00|
+-------------------+-------------------+
【讨论】：
我认为将时间戳四舍五入到最近日期的最简单方法是添加半天（12 小时），然后使用 date_trunc 将生成的时间戳截断为 DD 单位（天）。这将独立于您的数据存在的时间：
from pyspark.sql import functions as F
df1 = df.withColumn(
"output_column",
F.date_trunc("DD", F.col("datetime_column") + F.expr("INTERVAL 12 HOURS"))
)
df1.show(truncate=False)
#+-------------------+-------------------+
#|datetime_column |output_column |
#+-------------------+-------------------+
#|2021-01-29 21:30:00|2021-01-30 00:00:00|
#|2021-01-30 05:30:00|2021-01-30 00:00:00|
#|2021-01-01 23:25:00|2021-01-02 00:00:00|
#|2021-01-02 08:59:59|2021-01-02 00:00:00|
#+-------------------+-------------------+
使用普通的 Spark-SQL：
SELECT DATE_TRUNC('DD', datetime_column + INTERVAL 12 HOURS)
【讨论】：