使用 PySpark 将日期和时间字符串转换为时间戳时如何保留毫秒？

Posted 2023-02-24

【中文标题】使用 PySpark 将日期和时间字符串转换为时间戳时如何保留毫秒？

【英文标题】：How to preserve milliseconds when converting a date and time string to timestamp using PySpark?

【发布时间】：2019-04-17 16:21:27

【问题描述】：

我正在尝试将包含日期和时间作为字符串的列转换为时间戳，但是在转换过程中我丢失了毫秒部分。

数据

我有一个 Spark 数据框 df，它有一个包含字符串的日期和时间列。时间字符串包含毫秒，如下图：

+---------+------------+
|date     |time        |
+---------+------------+
|2018/1/2 |09:53:25.864|
|2018/1/3 |11:32:21.689|
|2018/1/4 |09:34:51.045|
+---------+------------+

我尝试了什么

我将date 和time 列连接起来得到date_and_time 列（字符串）：

import spark.sql.functions as F

df = df.withColumn('date_and_time', F.concat_ws(' ', df.date, df.time))

df.show(3, False)

输出：

+--------+------------+---------------------+
|date    |time        |date_and_time        |
+--------+------------+---------------------+
|2018/1/2|09:53:25.864|2018/1/2 09:53:25.864|
|2018/1/3|11:32:21.689|2018/1/3 11:32:21.689|
|2018/1/4|09:34:51.045|2018/1/4 09:34:51.045|
+--------+------------+---------------------+

然后，我使用Simple Date Format Date and Time patterns指定了时间戳格式：

timestamp_format = 'yyyy/M/d HH:mm:ss.SSS'

然后，我尝试使用几种不同的方式将此字符串转换为时间戳：

df.select(
    df.date_and_time,
    F.to_timestamp(df.date_and_time, timestamp_format).alias('method_1'),
    F.unix_timestamp(df.date_and_time, format=timestamp_format).cast('timestamp').alias('method_2')
).show(3, False)

如下所示，时间戳缺少毫秒部分：

+---------------------+-------------------+-------------------+
|date_and_time        |method_1           |method_2           |
+---------------------+-------------------+-------------------+
|2018/1/2 09:53:25.864|2018-01-02 09:53:25|2018-01-02 09:53:25|
|2018/1/3 11:32:21.689|2018-01-03 11:32:21|2018-01-03 11:32:21|
|2018/1/4 09:34:51.045|2018-01-04 09:34:51|2018-01-04 09:34:51|
+---------------------+-------------------+-------------------+

如何在将字符串转换为时间戳时保留毫秒？

我正在使用 PySpark（Spark：2.3.1，Python：3.6.5）。

我查看了之前回答的关于 SO 的问题，但没有找到合适的解决方案。

【问题讨论】：

这有帮助吗？ ***.com/a/54340652/4383754

【参考方案1】：

尽管这是一篇旧帖子，但我认为它可能对人们有用。 https://***.com/a/54340652/4383754 中的解决方案可能是最好的扩展方式。

如果您正在寻找一种更简单的解决方案，可以接受使用 python UDF 带来的性能损失，这里有一个：

from pyspark.sql.types import TimestampType
from pyspark.sql.functions import udf
from dateutil.parser import parse
data = [('2018/1/2', '09:53:25.864', '2018/1/2 09:53:25.864'),
        ('2018/1/3', '11:32:21.689', '2018/1/3 11:32:21.689'),
        ('2018/1/4', '09:34:51.045', '2018/1/4 09:34:51.045')]
df = spark.createDataFrame(
    data, 'date STRING, time STRING, date_and_time STRING')
parse_udf = udf(parse, TimestampType())
df = df.withColumn('parsed', parse_udf(df['date_and_time']))
df.show()
# +--------+------------+--------------------+--------------------+
# |    date|        time|       date_and_time|              parsed|
# +--------+------------+--------------------+--------------------+
# |2018/1/2|09:53:25.864|2018/1/2 09:53:25...|2018-01-02 09:53:...|
# |2018/1/3|11:32:21.689|2018/1/3 11:32:21...|2018-01-03 11:32:...|
# |2018/1/4|09:34:51.045|2018/1/4 09:34:51...|2018-01-04 09:34:...|
# +--------+------------+--------------------+--------------------+

df.dtypes
# [('date', 'string'),
#  ('time', 'string'),
#  ('date_and_time', 'string'),
#  ('parsed', 'timestamp')]

df[['parsed']].collect()[0][0]
# datetime.datetime(2018, 1, 2, 9, 53, 25, 864000) <- contains microsecond