Spark作业写入目标时字符串转Timestamp失败，报CAST_INVALID_INPUT错误求助

错误详情 #

作业因阶段失败中止：阶段15526.0中的任务18失败4次，最近一次失败：丢失阶段15526.0中的任务18.3（TID 3281950）（10.179.0.125执行器1190）：org.apache.spark.SparkDateTimeException: [CAST_INVALID_INPUT] 类型为“STRING”的值'04/12/2024 01:37:07.000 AM'无法转换为“TIMESTAMP”，因为格式不正确。请按照语法修正该值，或更改目标类型。

已尝试的解决方法 #

• 嵌套调用日期转换函数：

  df_new= (df.withColumn("Date",to_date( to_timestamp("LastUpdateDate","MM/dd/yyyy hh:mm:ss.SSS a"))))
  

• 使用try_cast表达式转换：

  dfnew = df.withColumn("Date", expr("try_cast(LastUpdateDateNew AS DATE)"))
  

• 通过unix_timestamp转换时间戳：

  # 将字符串转换为时间戳
  df_new = df.withColumn("LastUpdateTimestamp", unix_timestamp("LastUpdateDate", "MM/dd/yyyy            hh:mm:ss:SSS a").cast("timestamp"))
  
  # 将时间戳转换为MM/dd/YYYY格式的日期
  #df_new_bill = df_new.withColumn("date", to_date((col("LastUpdateTimestamp")), "MM/dd/yyyy"))
  

问题根源与解决办法 #

问题点 #

1. 第三个方法中格式字符串错误：原日期的毫秒部分用**点（.）分隔（如01:37:07.000），但代码里写成了冒号（:）**分隔（hh:mm:ss:SSS），导致格式匹配失败。
2. 未启用容错转换：直接使用to_timestamp会在遇到格式错误的数据时抛出异常中断作业，需要加入容错处理逻辑。

推荐解决方案 #

方案1：修正格式并使用容错转换

from pyspark.sql.functions import to_timestamp, to_date

# 容错转换为时间戳，使用正确的格式字符串
df_new = df.withColumn(
    "LastUpdateTimestamp",
    to_timestamp("LastUpdateDate", "MM/dd/yyyy hh:mm:ss.SSS a")
)
# 转换为日期类型
df_new = df_new.withColumn("Date", to_date("LastUpdateTimestamp"))

方案2：结合try_cast与格式转换

from pyspark.sql.functions import expr

df_new = df.withColumn(
    "Date",
    expr("try_cast(to_timestamp(LastUpdateDate, 'MM/dd/yyyy hh:mm:ss.SSS a') AS DATE)")
)

try_cast会在转换失败时返回null，不会中断整个作业。

方案3：提前过滤脏数据

如果数据中存在大量格式不规范的记录，可以先筛选出符合格式的内容再转换：

from pyspark.sql.functions import expr, to_date, to_timestamp

# 用正则筛选符合格式的日期字符串
valid_df = df.filter(
    expr("regexp_like(LastUpdateDate, '^\\d{2}/\\d{2}/\\d{4} \\d{2}:\\d{2}:\\d{2}\\.\\d{3} (AM|PM)$')")
)
# 对有效数据执行转换
valid_df = valid_df.withColumn(
    "Date",
    to_date(to_timestamp("LastUpdateDate", "MM/dd/yyyy hh:mm:ss.SSS a"))
)

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