PySpark：识别指定日期格式的字符串列并转换格式

原代码问题分析 #

1. 异常捕获无效：Spark的to_timestamp函数转换失败时不会抛出异常，只会返回null，因此原代码的try-except块无法区分有效/无效列。
2. 列筛选逻辑缺失：最终输出的是所有字符串列，没有筛选出真正符合目标时间格式的列。

步骤1：识别符合格式的字符串列 #

我们需要检查每个字符串列中是否存在（或全部是）符合yyyy-MM-dd'T'HH:mm:ss'Z'格式的值，以此筛选出目标列。

代码实现

from pyspark.sql import functions as F
from pyspark.sql.types import StringType

# 初始化测试DataFrame
df = spark.createDataFrame([
    (1, "2024-01-04T12:39:53Z", "other_value", "test1"),
    (2, "2023-02-15T09:24:36Z", "2024-01-04T12:39:53Z", "test2"),
], ["id", "column1", "column2", "column3"])

target_format = "yyyy-MM-dd'T'HH:mm:ss'Z'"

# 获取所有字符串类型列
string_columns = [col for col in df.columns if df.schema[col].dataType == StringType()]

# 筛选：列中至少有一个值符合时间格式
valid_date_columns = []
for col_name in string_columns:
    # 统计转换后非空的行数
    valid_count = df.filter(F.to_timestamp(F.col(col_name), target_format).isNotNull()).count()
    if valid_count > 0:
        valid_date_columns.append(col_name)

print("符合格式的列:", valid_date_columns)
# 输出：['column1', 'column2']

可选调整

如果要求列中所有值都必须符合格式，将判断条件改为：

if valid_count == df.count():

步骤2：动态转换为目标格式 #

将筛选出的列转换为dd/MM/yyyy H:mm:ss格式，可选择保留原始无效值或转为null。

方案1：无效值转为null

output_format = "dd/MM/yyyy H:mm:ss"

for col_name in valid_date_columns:
    df = df.withColumn(
        col_name,
        F.date_format(F.to_timestamp(F.col(col_name), target_format), output_format)
    )

df.show(truncate=False)

输出结果：

+---+-------------------+-------------------+------+
|id |column1            |column2            |column3|
+---+-------------------+-------------------+------+
|1  |04/01/2024 12:39:53|null               |test1 |
|2  |15/02/2023 09:24:36|04/01/2024 12:39:53|test2 |
+---+-------------------+-------------------+------+

方案2：保留原始无效值

使用when-otherwise逻辑，仅转换有效格式的值：

for col_name in valid_date_columns:
    df = df.withColumn(
        col_name,
        F.when(
            F.to_timestamp(F.col(col_name), target_format).isNotNull(),
            F.date_format(F.to_timestamp(F.col(col_name), target_format), output_format)
        ).otherwise(F.col(col_name))
    )

df.show(truncate=False)

输出结果：

+---+-------------------+-------------------+------+
|id |column1            |column2            |column3|
+---+-------------------+-------------------+------+
|1  |04/01/2024 12:39:53|other_value        |test1 |
|2  |15/02/2023 09:24:36|04/01/2024 12:39:53|test2 |
+---+-------------------+-------------------+------+

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