Python Pandas+Snowflake：数值转datetime方法及时间戳异常排查

原本将Pandas DataFrame中格式为YYYY-MM-DDTHH:MM:SSZ的字符串列转换为datetime类型，存入Snowflake时可正常显示为timestamp类型；但后续合并新旧表时，先将Snowflake现有表和新表统一转为字符串类型，合并去重后再把CREATIONTIME列转回datetime类型，存入Snowflake后该列变成了与原时间戳不匹配的数值（例如2023-07-20T14:43:42Z转为1,689,864,222,000,000，经确认是带6小时时区差的微秒级时间戳）。需明确异常原因，以及如何将该数值类型转回datetime类型。

附执行代码：

import pandas as pd
from snowflake.snowpark import Session

# session = create snowflake session
# newdf = new data into a new dataframe
 
existingdf = session.table(tablename).toPandas()

existingdf = existingdf.astype(str)
newdf = newdf.astype(str)

df = pd.concat([existingdf,newdf],ignore_index=True,sort=False)
df = df.loc[df.astype(str).drop_duplicates().index]

df["CREATIONTIME"] = pd.to_datetime(df["CREATIONTIME"], utc=True)

df = session.create_dataframe(df)
df.write.mode(action).save_as_table("TABLE_NAME")

1. 时区与类型转换的连锁问题：当Snowflake的timestamp数据通过toPandas()转为DataFrame后，再用astype(str)强制转字符串，原本的UTC timestamp会带上Snowflake默认时区（如UTC-6）的偏移信息。后续用pd.to_datetime(..., utc=True)解析时，Pandas会将带时区偏移的字符串转成datetime64[ns, UTC]类型，但内部存储的是从epoch开始的微秒数。
2. Snowpark类型映射偏差：使用session.create_dataframe(df)将Pandas的datetime64[ns, UTC]数据传入Snowpark时，Snowpark未正确识别为timestamp类型，而是将其解析为原始的微秒数值，加上时区转换的差值，最终存入Snowflake后变成了带时区差的微秒数。

一、将Snowflake中的数值转回datetime类型 #

方法1：Snowflake SQL直接转换 #

通过SQL把微秒数值转换为timestamp：

-- 先修改列类型为TIMESTAMP_NTZ
ALTER TABLE TABLE_NAME ALTER COLUMN CREATIONTIME SET DATA TYPE TIMESTAMP_NTZ;
-- 将微秒数转为UTC时间戳
UPDATE TABLE_NAME 
SET CREATIONTIME = TO_TIMESTAMP(CREATIONTIME / 1000000) AT TIME ZONE 'UTC';

方法2：通过Pandas重新处理 #

拉回数据到Pandas，转换后重新存入：

# 读取Snowflake中的数据
df = session.table("TABLE_NAME").toPandas()
# 将微秒数值转为UTC datetime
df["CREATIONTIME"] = pd.to_datetime(df["CREATIONTIME"], unit='us', utc=True)
# 重新写入Snowflake
df = session.create_dataframe(df)
df.write.mode("overwrite").save_as_table("TABLE_NAME")

二、避免后续出现该问题的优化方案 #

不要全局转字符串，仅针对类型不匹配的列单独处理，保留CREATIONTIME的datetime类型：

import pandas as pd
from snowflake.snowpark import Session
from snowflake.snowpark.types import TimestampType, StringType

# session = create snowflake session
# newdf = new data into a new dataframe
 
existingdf = session.table(tablename).toPandas()

# 仅对非时间列转字符串，保留CREATIONTIME的datetime类型
cols_to_str = [col for col in existingdf.columns if col != "CREATIONTIME"]
existingdf[cols_to_str] = existingdf[cols_to_str].astype(str)
newdf[cols_to_str] = newdf[cols_to_str].astype(str)

# 合并并去重
df = pd.concat([existingdf, newdf], ignore_index=True, sort=False)
df = df.drop_duplicates(subset=cols_to_str + ["CREATIONTIME"])

# 显式指定schema，确保时间列映射为Snowflake的timestamp
schema = []
for col in df.columns:
    if col == "CREATIONTIME":
        schema.append((col, TimestampType()))
    else:
        schema.append((col, StringType()))

df = session.create_dataframe(df, schema=schema)
df.write.mode(action).save_as_table("TABLE_NAME")

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