Pyspark from_unixtime (unix_timestamp) 不转换为时间戳

Question

我在 Python 2.7 中使用 Pyspark。我在字符串中有一个日期列（带毫秒），并想转换为时间戳

这是我迄今为止尝试过的

df = df.withColumn('end_time', from_unixtime(unix_timestamp(df.end_time, '%Y-%M-%d %H:%m:%S.%f')) )

printSchema() 显示 end_time: string (nullable = true)

当我使用时间戳作为变量类型时

Answer 1

尝试使用from_utc_timestamp：

from pyspark.sql.functions import from_utc_timestamp

df = df.withColumn('end_time', from_utc_timestamp(df.end_time, 'PST')) 

您需要为函数指定一个时区，在这种情况下我选择了PST

如果这不起作用，请给我们一个显示df.end_time的几行示例

Answer 2

使用格式化为字符串的时间戳创建示例数据框：

import pyspark.sql.functions as F
df = spark.createDataFrame([('22-Jul-2018 04:21:18.792 UTC', ),('23-Jul-2018 04:21:25.888 UTC',)], ['TIME'])
df.show(2,False)
df.printSchema()

输出：

+----------------------------+
|TIME                        |
+----------------------------+
|22-Jul-2018 04:21:18.792 UTC|
|23-Jul-2018 04:21:25.888 UTC|
+----------------------------+
root
|-- TIME: string (nullable = true)

将字符串时间格式（包括毫秒）转换为unix_timestamp(double)。 由于 unix_timestamp() 函数不包括毫秒，我们需要使用另一个简单的 hack 来添加它以包含毫秒。 使用子字符串方法 (start_position = -7, length_of_substring=3) 从字符串中提取毫秒并将毫秒单独添加到 unix_timestamp . （转换为子字符串以浮动添加）

df1 = df.withColumn("unix_timestamp",F.unix_timestamp(df.TIME,'dd-MMM-yyyy HH:mm:ss.SSS z') + F.substring(df.TIME,-7,3).cast('float')/1000)

在 Spark 中将 unix_timestamp(double) 转换为 timestamp 数据类型。

df2 = df1.withColumn("TimestampType",F.to_timestamp(df1["unix_timestamp"]))
df2.show(n=2,truncate=False)

这将为您提供以下输出

+----------------------------+----------------+-----------------------+
|TIME                        |unix_timestamp  |TimestampType          |
+----------------------------+----------------+-----------------------+
|22-Jul-2018 04:21:18.792 UTC|1.532233278792E9|2018-07-22 04:21:18.792|
|23-Jul-2018 04:21:25.888 UTC|1.532319685888E9|2018-07-23 04:21:25.888|
+----------------------------+----------------+-----------------------+

检查架构：

df2.printSchema()


root
 |-- TIME: string (nullable = true)
 |-- unix_timestamp: double (nullable = true)
 |-- TimestampType: timestamp (nullable = true)

Answer 3

在当前版本的 spark 中，我们不必在时间戳转换方面做太多事情。

在这种情况下，使用 to_timestamp 函数效果很好。我们唯一需要注意的是根据原始列输入时间戳的格式。 就我而言，它的格式为 yyyy-MM-dd HH:mm:ss。 其他格式可以是 MM/dd/yyyy HH:mm:ss 或这样的组合。

from pyspark.sql.functions import to_timestamp
df=df.withColumn('date_time',to_timestamp('event_time','yyyy-MM-dd HH:mm:ss'))
df.show()

Answer 4

以下可能会有所帮助：-

from pyspark.sql import functions as F
df = df.withColumn("end_time", F.from_unixtime(F.col("end_time"), 'yyyy-MM-dd HH:mm:ss.SS').cast("timestamp"))

[更新]