pyspark to_timestamp 不包括毫秒答案

【问题标题】：pyspark to_timestamp does not include millisecondspyspark to_timestamp 不包括毫秒
【发布时间】：2019-01-04 18:06:58
【问题描述】：

我正在尝试格式化我的时间戳列以包含毫秒但没有成功。如何格式化我的时间看起来像这样 - 2019-01-04 11:09:21.152 ？

我查看了文档并遵循了 SimpleDataTimeFormat ，pyspark 文档说 to_timestamp 函数正在使用它。

这是我的数据框。

+--------------------------+
|updated_date              |
+--------------------------+
|2019-01-04 11:09:21.152815|
+--------------------------+

我使用毫秒格式没有任何成功，如下所示

>>> df.select('updated_date').withColumn("updated_date_col2", 
to_timestamp("updated_date", "YYYY-MM-dd HH:mm:ss:SSS")).show(1,False)
+--------------------------+-------------------+
|updated_date              |updated_date_col2  |
+--------------------------+-------------------+
|2019-01-04 11:09:21.152815|2019-01-04 11:09:21|
+--------------------------+-------------------+

我希望updated_date_col2 被格式化为2019-01-04 11:09:21.152

【问题讨论】：

时间戳类型没有毫秒。如果需要，您必须将列保留为字符串。

标签： apache-spark pyspark

【解决方案1】：

这不是 to_timestamp 的解决方案，但您可以轻松地将列保持为时间格式

以下代码是将数值毫秒转换为时间戳的示例之一。

from datetime import datetime

ms = datetime.now().timestamp() # ex) ms = 1547521021.83301
df = spark.createDataFrame([(1, ms)], ['obs', 'time'])
df = df.withColumn('time', df.time.cast("timestamp"))
df.show(1, False) 

+---+--------------------------+
|obs|time                      |
+---+--------------------------+
|1  |2019-01-15 12:15:49.565263|
+---+--------------------------+

如果你在 JS 中使用 new Date().getTime() 或 Date.now() 或在 Python 中使用 datetime.datetime.now().timestamp()，你可以获得一个数值毫秒。

【讨论】：

上述解决方案不起作用。即使我导入了日期时间库，我仍然收到'DataFrame' object has no attribute 'time'

【解决方案2】：

原因 pyspark to_timestamp 只解析到秒，而 TimestampType 有能力保持毫秒。

以下解决方法可能有效：

如果时间戳模式包含 S，则调用 UDF 以获取字符串 'INTERVAL MILLISECONDS' 以在表达式中使用

ts_pattern = "YYYY-MM-dd HH:mm:ss:SSS"
my_col_name = "time_with_ms"

# get the time till seconds
df = df.withColumn(my_col_name, to_timestamp(df["updated_date_col2"],ts_pattern))

# add milliseconds as inteval
if 'S' in timestamp_pattern:
   df = df.withColumn(my_col_name, df[my_col_name] + expr("INTERVAL 256 MILLISECONDS"))

要获得 INTERVAL 256 MILLISECONDS，我们可以使用 Java UDF：

df = df.withColumn(col_name, df[col_name] + expr(getIntervalStringUDF(df[my_col_name], ts_pattern)))

UDF 内部：getIntervalStringUDF(String timeString, String pattern)

使用 SimpleDateFormat 根据模式解析日期
使用模式“'INTERVAL 'SSS' MILLISECONDS'”将格式化日期作为字符串返回
在解析/格式异常时返回“INTERVAL 0 MILLISECONDS”

【讨论】：

【解决方案3】：

我认为您可以使用 UDF 和 Python 的标准日期时间模块，如下所示。

import datetime
from pyspark.sql.functions import udf
from pyspark.sql.types import TimestampType

def _to_timestamp(s):
    return datetime.datetime.strptime(s, '%Y-%m-%d %H:%M:%S.%f')

udf_to_timestamp = udf(_to_timestamp, TimestampType())

df.select('updated_date').withColumn("updated_date_col2", udf_to_timestamp("updated_date")).show(1,False)

【讨论】：