Databricks PySpark 实现Quickbase毫秒时间戳列的标准日期格式转换

• 存在两个关键逻辑缺陷：
  • PySpark中col()返回的是分布式列引用对象，不属于Python原生数值类型，直接将该对象传入datetime.datetime.fromtimestamp()这类Python单值处理方法时，方法收到不符合预期的列类型，直接触发“Expecting integer but received col type”报错。这类Python原生方法本身不具备分布式列逐行处理的能力。
  • 原始字段存储的是13位毫秒级时间戳，数值远超过32位int类型的存储上限，强转int的操作会导致数据截断，无法得到正确计算结果。必须使用bigint类型承接毫秒值转换，否则会出现溢出截断。

PySpark自带的日期处理函数运行在Spark执行引擎层，不需要序列化Python计算逻辑，分布式场景下的处理效率远高于逐行Python调用。
实现逻辑拆解：

1. 将原始string类型的毫秒时间戳转换为bigint类型，规避int溢出问题
2. 将毫秒值除以1000转为秒级Unix时间戳
3. 调用from_unixtime函数直接格式化为yyyy-MM-dd格式的日期字符串

对应代码：

from pyspark.sql.functions import col, from_unixtime

df_transformed = (
    df
    .withColumn(
        "product_availability_due_date_ms",
        col("product_availability_due_date").cast("bigint")
    )
    .withColumn(
        "product_availability_due_date_1",
        from_unixtime(
            col("product_availability_due_date_ms") / 1000,
            "yyyy-MM-dd"
        )
    )
)

如果必须复用原生datetime的转换逻辑，需要通过pandas_udf做批量适配，严禁直接将列对象传入Python单值方法：

import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType

@pandas_udf(StringType())
def ms_to_date(ms_series: pd.Series) -> pd.Series:
    # 批量处理整列毫秒值，转换效果和单值Python逻辑完全一致
    return pd.to_datetime(ms_series / 1000, unit="ms").dt.strftime("%Y-%m-%d")

# 应用列转换
df_transformed = df.withColumn(
    "product_availability_due_date_1",
    ms_to_date(col("product_availability_due_date").cast("bigint"))
)

• 转换过程中遇到非数字、空值等脏数据时，Spark会自动将对应值转为null，不会中断任务执行。
• 不建议使用普通逐行Python UDF实现转换，这类UDF需要逐行序列化Python计算逻辑，大数据量场景下性能损耗会非常明显。

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