PySpark 数据框将不寻常的字符串格式转换为时间戳

Question

我通过 Spark 1.5.0 使用 PySpark。 我在日期时间值的列的行中有一个不寻常的字符串格式。看起来像这样：

Row[(datetime='2016_08_21 11_31_08')]

有没有办法将这种非正统的

yyyy_mm_dd hh_mm_dd

格式转换为时间戳？ 最终可能会出现以下情况：

df = df.withColumn("date_time",df.datetime.astype('Timestamp'))

我原以为像

regexp_replace

这样的Spark SQL函数可以工作，但我当然需要替换

_

与

-

位于日期的一半 和

_

以及时间部分中的

:

。

我想我可以使用

substring

将列分成 2 部分，并从时间结束时向后计数。然后分别执行“regexp_replace”，然后连接。不过这样操作好像很多吧？有没有更简单的方法？

Answer 1

火花 >= 2.2

from pyspark.sql.functions import to_timestamp

(sc
    .parallelize([Row(dt='2016_08_21 11_31_08')])
    .toDF()
    .withColumn("parsed", to_timestamp("dt", "yyyy_MM_dd HH_mm_ss"))
    .show(1, False))

## +-------------------+-------------------+
## |dt                 |parsed             |
## +-------------------+-------------------+
## |2016_08_21 11_31_08|2016-08-21 11:31:08|
## +-------------------+-------------------+

火花< 2.2

没有什么是

unix_timestamp

解决不了的：

from pyspark.sql import Row
from pyspark.sql.functions import unix_timestamp

(sc
    .parallelize([Row(dt='2016_08_21 11_31_08')])
    .toDF()
    .withColumn("parsed", unix_timestamp("dt", "yyyy_MM_dd HH_mm_ss")
    # For Spark <= 1.5
    # See issues.apache.org/jira/browse/SPARK-11724 
    .cast("double")
    .cast("timestamp"))
    .show(1, False))

## +-------------------+---------------------+
## |dt                 |parsed               |
## +-------------------+---------------------+
## |2016_08_21 11_31_08|2016-08-21 11:31:08.0|
## +-------------------+---------------------+

在这两种情况下，格式字符串都应与 Java

SimpleDateFormat

兼容。

Answer 2

zero323的答案回答了问题，但我想补充一点，如果您的日期时间字符串具有标准格式，您应该能够将其直接转换为时间戳类型：

df.withColumn('datetime', col('datetime_str').cast('timestamp'))

它的优点是可以处理毫秒，而unix_timestamp只有第二精度（to_timestamp也可以处理毫秒，但需要Spark >= 2.2，如zero323所述）。我在 Spark 2.3.0 上测试了它，使用以下格式：'2016-07-13 14:33:53.979'（有毫秒，但没有毫秒也能工作）。

Answer 3

我从 Florent F 的回答添加了更多代码行，以便更好地理解和在本地计算机中运行代码片段：

import os, pdb, sys
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark.sql.types import StructType, ArrayType  
from pyspark.sql.types import StringType
from pyspark.sql.functions import col

sc = pyspark.SparkContext('local[*]')
spark = SparkSession.builder.getOrCreate()

# preparing some example data - df1 with String type and df2 with Timestamp type
df1 = sc.parallelize([{"key":"a", "date":"2016-02-01"}, 
    {"key":"b", "date":"2016-02-02"}]).toDF()
df1.show()

df2 = df1.withColumn('datetime', col('date').cast("timestamp"))
df2.show()

Answer 4

只是想在本次讨论中添加更多资源和示例。 https://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html 例如，如果您的 ts 字符串是“22 Dec 2022 19:06:36 EST”，则格式为“dd MMM yyyy HH:mm:ss zzz”

Answer 5

+1

火花 - 2.4：

to_timestamp() 无法处理夏令时的日期时间字符串。unix_timestamp() 保存了它。我使用 from_utc_timestamp() 进行区域转换。

直接使用日期时间字符串通过 from_utc_timestamp() 给出 1 小时的错误结果。通过使用 from_utc_timestamp(unix_timestamp().cast(TimestampType()) ) 给出了正确的转换。