PySpark字符串转yyyy-MM-dd HH:mm:ss格式Timestamp失败求助

你的代码运行失败，核心原因大概率是合并后的字符串不符合yyyy-MM-dd HH:mm:ss格式，或是数据里存在空值、非标准时间格式等异常值，导致to_timestamp转换报错。以下是具体的排查和修复步骤：

一、先定位问题根源 #

1. 验证日期列格式
   先确认CurrentStateDate是否真的是yyyy-MM-dd格式，执行代码查看样本数据：

   df.select("CurrentStateDate").show(10, truncate=False)
   

   如果是其他格式（比如dd/MM/yyyy），直接拼接后必然不符合转换要求。

2. 检查时间列异常值
   查看CurrentStateTime处理后是否存在无效值，比如空值、长度非4位的内容：

   from pyspark.sql import functions as F
   df.select("CurrentStateTime")\
     .filter(F.col("CurrentStateTime").isNull() | (F.length(F.regexp_replace(F.col("CurrentStateTime"), " ", "")) !=4))\
     .show(10, truncate=False)
   

   比如原始时间是123，去空格后变成123，拼接后会生成yyyy-MM-dd 123:00，这显然不是合法的HH:mm格式，直接导致转换失败。

二、针对性修复方案 #

方案1：标准化日期+清洗时间后转换 #

先把日期列统一为yyyy-MM-dd格式，同时过滤/修复时间列的异常值：

from pyspark.sql import functions as F

# 1. 标准化日期列（如果原始格式不是yyyy-MM-dd，替换成实际格式，比如"dd/MM/yyyy"）
df = df.withColumn("CleanedDate", F.date_format(F.to_date(F.col("CurrentStateDate"), "yyyy-MM-dd"), "yyyy-MM-dd"))

# 2. 清洗时间列：仅保留有效4位时间，异常值设为0000（可根据业务调整规则）
df = df.withColumn("CleanedTime", F.when(
    F.length(F.regexp_replace(F.col("CurrentStateTime"), " ", "")) ==4,
    F.regexp_replace(F.col("CurrentStateTime"), " ", "")
).otherwise(F.lit("0000")))

# 3. 拆分时间为时分，拼接成标准格式后转Timestamp
df = df.withColumn("CurrentStateDateTime", F.to_timestamp(
    F.concat(
        F.col("CleanedDate"),
        F.lit(" "),
        F.substring(F.col("CleanedTime"), 1, 2),
        F.lit(":"),
        F.substring(F.col("CleanedTime"), 3, 2),
        F.lit(":00")
    ),
    "yyyy-MM-dd HH:mm:ss"
))

# 清理中间临时列
df = df.drop("CleanedDate", "CleanedTime")

方案2：用时间戳数值计算，避免字符串拼接 #

如果日期列是date类型、时间列是HHmm格式，可以用时间戳数值相加的方式生成Timestamp，稳定性更高：

from pyspark.sql import functions as F

# 将日期转成秒级时间戳
df = df.withColumn("DateTimestamp", F.unix_timestamp(F.col("CurrentStateDate"), "yyyy-MM-dd"))

# 将HHmm格式的时间转成秒数
df = df.withColumn("TimeSeconds", F.unix_timestamp(F.regexp_replace(F.col("CurrentStateTime"), " ", ""), "HHmm"))

# 时间戳相加后转成标准Timestamp
df = df.withColumn("CurrentStateDateTime", F.to_timestamp(F.col("DateTimestamp") + F.col("TimeSeconds")))

# 清理中间临时列
df = df.drop("DateTimestamp", "TimeSeconds")

方案3：添加容错逻辑处理异常值 #

如果数据里存在无法修复的异常值，用coalesce避免转换失败：

df = df.withColumn("CurrentStateDateTime", F.coalesce(
    F.to_timestamp(F.concat(...), "yyyy-MM-dd HH:mm:ss"),  # 你的正常转换逻辑
    F.lit(None)  # 异常值设为Null，也可根据业务设默认时间如"1970-01-01 00:00:00"
))

三、调试技巧 #

在流水线中加入中间结果查看步骤，快速定位问题数据：

df.select("CurrentStateDate", "CurrentStateTime", "CurrentStateDateTime").show(20, truncate=False)

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