Snowflake TIMESTAMP_NTZ无效日期问题：ADF复制活动转换失败

ADF端处理方案 #

1. 用数据流动（Data Flow）实现转换与过滤 #

这是最灵活的方案，能在数据加载过程中直接处理无效日期：

• 源配置：将Snowflake源中的REFERENCE_DATETIME列以字符串类型读取（在源数据集的"连接"或数据流动的源转换中，手动指定列类型为String），避免ADF自动尝试转换DateTime报错。
• 派生列转换：添加派生列，用try()函数尝试转换日期，无效值转为null：

  try(toTimestamp(REFERENCE_DATETIME, 'yyyy-MM-dd HH:mm:ss.SSS'))
  

  也可以指定默认值替代null，比如：

  iif(isValid(REFERENCE_DATETIME, 'yyyy-MM-dd HH:mm:ss.SSS'), toTimestamp(REFERENCE_DATETIME), toTimestamp('1900-01-01 00:00:00'))
  

• 筛选转换：如果需要直接过滤掉无效日期的行，添加筛选转换，条件设为：

  isValid(REFERENCE_DATETIME, 'yyyy-MM-dd HH:mm:ss.SSS')
  

• 目标输出：将处理后的列映射到Parquet目标数据集，正常输出。

2. Copy Activity临时容错（不推荐） #

如果不想用数据流动，可临时开启跳过错误行功能，但会丢失无效数据：

• 在Copy Activity的设置标签页，找到容错选项，开启"跳过错误行"，设置允许跳过的最大行数。

Databricks端后处理方案 #

如果已经将带无效日期的数据加载到Parquet，可在Databricks中进行清洗：

1. 读取并转换无效日期 #

先以字符串类型读取列，再用try_cast转换为Timestamp，无效值自动转为null：

from pyspark.sql.types import StringType, TimestampType
from pyspark.sql.functions import col, try_cast

# 读取Parquet文件，强制将目标列转为字符串
df = spark.read.parquet("/path/to/your/parquet") \
    .withColumn("REFERENCE_DATETIME", col("REFERENCE_DATETIME").cast(StringType()))

# 尝试转换为Timestamp，无效值转null
df = df.withColumn("valid_reference_datetime", try_cast(col("REFERENCE_DATETIME"), TimestampType()))

2. 过滤无效行 #

过滤掉转换后为null的行：

clean_df = df.filter(col("valid_reference_datetime").isNotNull())

3. 替换无效值为默认值 #

如果不想丢弃行，可将无效日期替换为指定默认值：

from pyspark.sql.functions import when, lit, isValid

df = df.withColumn(
    "valid_reference_datetime",
    when(
        isValid(col("REFERENCE_DATETIME"), "yyyy-MM-dd HH:mm:ss.SSS"),
        col("REFERENCE_DATETIME").cast(TimestampType())
    ).otherwise(
        lit("1900-01-01 00:00:00").cast(TimestampType())
    )
)

源端预处理（Snowflake） #

从源头过滤无效数据是最高效的方式，在Snowflake的源查询中直接处理：

SELECT *
FROM your_source_table
WHERE TRY_TO_TIMESTAMP(REFERENCE_DATETIME) IS NOT NULL

这样ADF读取的就是已经过滤好的有效数据，避免后续转换报错。

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