如何在PySpark SQL Schema中定义日期格式以解析为Timestamp

你完全可以省去后续的DataFrame转换步骤，直接在读取CSV时就指定日期格式，让Spark自动将字符串解析为TimestampType。下面是具体的实现方法：

直接解析的解决方案 #

首先，在定义Schema时直接将日期字段指定为TimestampType，然后在读取CSV时通过timestampFormat参数匹配你的输入日期格式：

from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

# 直接将日期字段定义为TimestampType，无需先设为StringType
my_schema = StructType([
    StructField('my_integer', IntegerType()),
    StructField('my_string', StringType()),
    StructField('my_date', TimestampType())
])

# 读取CSV时指定timestampFormat，匹配你的输入格式"dd/MM/yyyy HH:mm:ss"
df_test = spark.read.schema(my_schema) \
    .option("timestampFormat", "dd/MM/yyyy HH:mm:ss") \
    .csv('my_data.csv')

验证结果 #

执行后查看数据和Schema：

df_test.show()

输出：

+----------+---------+-------------------+
|my_integer|my_string|           my_date|
+----------+---------+-------------------+
|         1|      Foo|2017-06-19 14:41:20|
|         2|      Bar|2018-06-19 15:41:45|
+----------+---------+-------------------+

df_test.printSchema()

输出：

root
 |-- my_integer: integer (nullable = true)
 |-- my_string: string (nullable = true)
 |-- my_date: timestamp (nullable = true)

原理说明 #

PySpark的CSV数据源内置了对日期/时间戳格式的解析支持：

• 通过timestampFormat参数，你可以告诉Spark输入字符串的时间戳格式
• 当Schema中字段类型为TimestampType时，Spark会自动按照指定格式解析字符串，直接转换为Timestamp类型，无需额外的to_timestamp转换

注意事项 #

• 格式字符串的大小写要严格匹配：比如MM代表月份（小写mm是分钟），HH是24小时制小时（小写hh是12小时制），写错会导致解析失败，字段值变为null
• 该方法适用于Spark 2.2及以上版本（如果你的版本过低，可能需要升级或者 fallback 到你原来的间接方法）
• 如果CSV中有多种不同格式的时间戳字段，这种统一指定格式的方法就不适用了，此时还是需要用你原来的方法，针对每个字段单独调用to_timestamp转换

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