在 PySpark 中使用微秒时间戳

Question

我有一个时间格式为20190111-08:15:45.275753的pyspark数据帧。我想将其转换为保持微秒粒度的时间戳格式。但是，似乎很难保持微秒，因为pyspark中的所有时间转换都会产生秒？

您知道如何做到这一点吗？请注意，将其转换为熊猫等将不起作用，因为数据集很大，所以我需要一种有效的方法来做到这一点。下面我如何执行此操作的示例

time_df = spark.createDataFrame([('20150408-01:12:04.275753',)], ['dt'])
res = time_df.withColumn("time",  unix_timestamp(col("dt"), 
format='yyyyMMdd-HH:mm:ss.000').alias("time"))
res.show(5, False)

Answer 1

通常时间戳粒度以秒为单位，因此我认为没有直接的方法可以保持毫秒粒度。

在pyspark中有一个函数unix_timestamp：

unix_timestamp(timestamp=None, format='yyyy-MM-dd HH:mm:ss')

使用给定模式转换时间字符串（默认为'yyyy-MM-dd HH:mm:ss'） 到 Unix 时间戳（以秒为单位），使用默认时区和默认时区 区域设置，如果失败，则返回 null。

if `timestamp` is None, then it returns current timestamp.
>>> spark.conf.set("spark.sql.session.timeZone", "America/Los_Angeles")
>>> time_df = spark.createDataFrame([('2015-04-08',)], ['dt'])
>>> time_df.select(unix_timestamp('dt', 'yyyy-MM-dd').alias('unix_time')).collect()
[Row(unix_time=1428476400)]
>>> spark.conf.unset("spark.sql.session.timeZone")

使用示例：

import pyspark.sql.functions as F
res = df.withColumn(colName,  F.unix_timestamp(F.col(colName), 
    format='yyyy-MM-dd HH:mm:ss.000').alias(colName) )

您可以做的是拆分日期字符串 （ str.rsplit('.', 1) ），在数据帧中保持毫秒间隔（例如，通过创建另一列）。

编辑

在您的示例中，问题是时间的类型是字符串。首先，您需要将其转换为timestamp类型：这可以通过以下方式完成：

res = time_df.withColumn("new_col", to_timestamp("dt", "yyyyMMdd-hh:mm:ss"))

然后你可以使用unix_timestamp

res2 = res.withColumn("time",  F.unix_timestamp(F.col("parsed"), format='yyyyMMdd-hh:mm:ss.000').alias("time"))

最后创建一个毫秒的列：

res3 = res2.withColumn("ms", F.split(res2['dt'], '[.]').getItem(1))

Answer 2

我已经在pyspark中使用to_utc_timestamp函数找到了解决此问题的方法，但是不完全确定这是否是最有效的，尽管它似乎在大约100 mn行数据上工作正常。如果您的时间戳字符串如下所示，您可以避免regex_replace -1997-02-28 10：30：40.897748

 from pyspark.sql.functions import regexp_replace, to_utc_timestamp
 df = spark.createDataFrame([('19970228-10:30:40.897748',)], ['new_t'])
 df = df.withColumn('t', regexp_replace('new_t', '^(.{4})(.{2})(.{2})-', '$1-$2-$3 '))
 df = df.withColumn("time", to_utc_timestamp(df.t, "UTC").alias('t'))
 df.show(5,False)
 print(df.dtypes)