如何在PySpark DataFrame中提取微秒值？

问题背景 #

现有如下PySpark DataFrame：

from datetime import datetime
data = [
    (1, datetime(2018, 7, 25, 17, 15, 6, 390000)),
    (2, datetime(2018, 7, 25, 11, 12, 49, 317000))
]
df = spark.createDataFrame(data, ['ID', 'max_ts'])

DataFrame内容：

+---+-----------------------+
|ID |max_ts                 |
+---+-----------------------+
|1  |2018-07-25 17:15:06.39 |
|2  |2018-07-25 11:12:49.317|
+---+-----------------------+

需求是新增一列ms，存储max_ts字段的微秒值，目标结果：

+---+-----------------------+------+
|ID |max_ts                 |ms    |
+---+-----------------------+------+
|1  |2018-07-25 17:15:06.39 |390000|
|2  |2018-07-25 11:12:49.317|317000|
+---+-----------------------+------+

在Pandas中可以通过如下代码实现：

df_interfax['ms_created_at'] = df_interfax['max_ts'].dt.microsecond

请问在PySpark中该如何实现这一需求？

解决方案 #

可以通过两种方式实现：

方法一：使用date_format结合字符串处理 #

利用date_format提取微秒部分的字符串，转换为整数即可，该函数会自动补全不足6位的微秒值：

from pyspark.sql.functions import date_format, col

df_result = df.withColumn(
    "ms",
    date_format(col("max_ts"), "SSSSSS").cast("int")
)
df_result.show(truncate=False)

方法二：通过时间戳差值计算 #

计算完整时间戳与截断到秒的时间戳差值，再乘以1000000得到微秒数：

from pyspark.sql.functions import unix_timestamp, col

df_result = df.withColumn(
    "ms",
    (unix_timestamp(col("max_ts"), "yyyy-MM-dd HH:mm:ss.SSSSSS") - unix_timestamp(col("max_ts"), "yyyy-MM-dd HH:mm:ss")) * 1000000
)
df_result.show(truncate=False)

说明 #

• 方法一中，date_format的SSSSSS格式符会直接提取6位微秒值，对于显示为.39的时间，会自动补全为390000，完全匹配需求。
• 方法二中，通过两个时间戳的差值拿到秒级以下的小数部分，放大1000000倍后得到精确的微秒数，结果一致。

内容的提问来源于stack exchange，提问作者Rory