如何高效处理CSV大数据中含Unixtimestamp与normal Timestamp的日期列并统一格式？

刚好之前处理过类似的混合日期格式场景，给你分享两种高效的解决方案——SQL查询和PySpark DataFrame处理，都是针对大数据量优化过的，放心用：

核心思路是通过CASE WHEN分支判断，先区分字段是Unix时间戳（通常是10位秒级或13位毫秒级数字）还是普通时间戳字符串，再调用对应数据库的日期转换函数统一格式。

下面是主流数据库的示例代码（假设日期列名为date_col，表名为big_data_table）：

MySQL 示例 #

-- 查询转换后的统一日期
SELECT
    CASE
        -- 匹配10位秒级Unix时间戳
        WHEN date_col REGEXP '^[0-9]{10}$' THEN FROM_UNIXTIME(CAST(date_col AS UNSIGNED))
        -- 匹配13位毫秒级Unix时间戳，需要除以1000转成秒
        WHEN date_col REGEXP '^[0-9]{13}$' THEN FROM_UNIXTIME(CAST(date_col AS UNSIGNED)/1000)
        -- 处理普通时间戳字符串（假设格式为'YYYY-MM-DD HH:MM:SS'，可根据实际格式调整）
        ELSE STR_TO_DATE(date_col, '%Y-%m-%d %H:%i:%s')
    END AS unified_date
FROM big_data_table;

-- 如果需要将转换结果持久化到表中
ALTER TABLE big_data_table ADD COLUMN unified_date DATETIME;
UPDATE big_data_table
SET unified_date = CASE
    WHEN date_col REGEXP '^[0-9]{10}$' THEN FROM_UNIXTIME(CAST(date_col AS UNSIGNED))
    WHEN date_col REGEXP '^[0-9]{13}$' THEN FROM_UNIXTIME(CAST(date_col AS UNSIGNED)/1000)
    ELSE STR_TO_DATE(date_col, '%Y-%m-%d %H:%i:%s')
END;

PostgreSQL 示例 #

PostgreSQL的函数略有不同，用TO_TIMESTAMP替代FROM_UNIXTIME：

SELECT
    CASE
        WHEN date_col ~ '^[0-9]{10}$' THEN TO_TIMESTAMP(CAST(date_col AS BIGINT))
        WHEN date_col ~ '^[0-9]{13}$' THEN TO_TIMESTAMP(CAST(date_col AS BIGINT)/1000)
        ELSE TO_TIMESTAMP(date_col, 'YYYY-MM-DD HH24:MI:SS')
    END AS unified_date
FROM big_data_table;

注意：如果你的普通时间戳有多种格式，可以在ELSE分支里嵌套多个STR_TO_DATE/TO_TIMESTAMP尝试，或者用数据库的自动类型转换（比如PostgreSQL的TRY_CAST）。

PySpark适合处理超大规模数据，核心是用内置的分布式函数（避免慢的UDF）来做分支转换，效率很高。

基础转换示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, from_unixtime, to_timestamp

# 初始化SparkSession
spark = SparkSession.builder.appName("DateUnification").getOrCreate()

# 读取CSV文件（inferSchema=True会自动推断列类型，也可以手动指定）
df = spark.read.csv("your_big_data.csv", header=True, inferSchema=True)

# 生成统一格式的日期列（这里统一转成timestamp类型，也可以指定字符串格式）
unified_df = df.withColumn(
    "unified_date",
    when(
        # 匹配10位秒级Unix时间戳
        col("date_col").rlike("^[0-9]{10}$"),
        from_unixtime(col("date_col").cast("bigint")).cast("timestamp")
    ).when(
        # 匹配13位毫秒级Unix时间戳
        col("date_col").rlike("^[0-9]{13}$"),
        from_unixtime(col("date_col").cast("bigint")/1000).cast("timestamp")
    ).otherwise(
        # 处理普通时间戳字符串，自动识别常见格式
        to_timestamp(col("date_col"))
    )
)

# 查看结果或保存
unified_df.show()
# unified_df.write.mode("overwrite").parquet("unified_data.parquet")

更灵活的格式兼容方案 #

如果普通时间戳有多种不固定格式，可以用try_cast自动尝试转换，不用手动指定格式：

from pyspark.sql.functions import try_cast

unified_df = df.withColumn(
    "unified_date",
    when(
        col("date_col").rlike("^[0-9]{10}$"),
        from_unixtime(col("date_col").cast("bigint")).cast("timestamp")
    ).when(
        col("date_col").rlike("^[0-9]{13}$"),
        from_unixtime(col("date_col").cast("bigint")/1000).cast("timestamp")
    ).otherwise(
        try_cast(col("date_col").cast("string"), "timestamp")
    )
)

优化提示：大数据量下尽量避免使用自定义UDF，PySpark内置函数是经过分布式优化的，性能会好很多；如果日期列的类型混合（部分是数字、部分是字符串），可以先转成字符串再判断。

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