PySpark DataFrame通过Sqoop upsert到SQL Server遇datetime2报错如何解决

你遇到的datetime2取值越界错误主要来自三个核心问题：

1. Parquet存储格式不识别你指定的timestampFormat参数，该参数仅对JSON、CSV等文本存储格式生效，Parquet为二进制存储，时间戳会按Spark内部的timestamp类型直接序列化，你指定的时间格式参数完全未生效。
2. 你写Parquet时用的时间格式配置本身有两个错误：YYYY代表的是ISO周年而非公历年，容易出现跨年时间错位；hh为12小时制标识，会导致下午的时间解析出现12小时偏差，超出SQL Server datetime2的合法范围。
3. Sqoop默认解析Parquet时间戳时会将INT96类型的时间戳转成错误的时间值，Spark默认写Parquet的时间戳用的就是INT96格式，进一步加剧了时间解析异常。

如果你要保留现有HDFS+Sqoop的链路，按以下步骤调整即可：

• 修改Spark写Parquet的逻辑，修正时间配置，指定时间戳输出格式为微秒级，避免INT96类型：

def write_df_to_hdfs(df, filename, hdfs_location_working):
        """
        Function to write delta records dataframe to HDFS
        """
        logging.info("Started writing delta records dataframe to hdfs")
        # 新增配置指定时间戳输出类型，移除对parquet无效的timestampFormat参数
        df.write.option("parquet.outputTimestampType", "TIMESTAMP_MICROS")\
                .save(hdfs_location_working, format='parquet', mode='append', emptyValue="")
        logging.info("Successfully written delta records dataframe to hdfs")

• 调整Sqoop命令，新增时间列映射规则，删除重复的--input-null-string参数：

sqoop export -Dmapreduce.map.memory.mb=4096 -Dmapreduce.map.java.opts=-Xmx3000m -Dmapred.job.queuename=ici -Dsqoop.export.records.per.statement=30 -Dsqoop.export.statements.per.transaction=30 -libjars /opt/cloudera/parcels/CDH-7.1.6-1.cdh7.1.6.p6.12486751/lib/sqoop/lib/sqljdbc.jar --connect "jdbc:sqlserver://*******.hosts.cloud.ford.com;databaseName=SQTDIAPM_AM;schema=dbo;" \
--username 'user' \
--password 'pwd' \
--export-dir <HDFS Path> \
--table <tablename> \
--input-null-string '\\N' \
--input-null-non-string '\\N' \
--update-key col1,col2 \
--update-mode allowinsert \
--map-column-java 你的时间列名1=java.sql.Timestamp,你的时间列名2=java.sql.Timestamp \
--batch \
-m 40 \
--verbose

无需绕HDFS和Sqoop，直接通过Spark JDBC结合SQL Server MERGE语句实现Upsert，稳定性更高，链路更短：

1. 首先对齐DataFrame列类型和SQL Server目标表的列类型，特别是datetime2类型的列要先转换为Spark的TimestampType，避免隐式转换异常。
2. 实现临时表+MERGE的Upsert逻辑，代码示例如下：

import uuid
from py4j.java_gateway import java_import

# JDBC连接配置
jdbc_url = "jdbc:sqlserver://*******.hosts.cloud.ford.com;databaseName=SQTDIAPM_AM;schema=dbo;"
conn_props = {
    "user": "user",
    "password": "pwd",
    "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver",
    "batchsize": "1000",
    "rewriteBatchedStatements": "true"
}
target_table = "你的目标表名"
# 生成唯一临时表名避免冲突
tmp_table = f"tmp_upsert_{str(uuid.uuid4()).replace('-', '')}"

# 1. 将增量数据写入SQL Server临时表
df.write.jdbc(
    url=jdbc_url,
    table=tmp_table,
    mode="overwrite",
    properties=conn_props
)

# 2. 执行MERGE语句实现Upsert
java_import(spark._jvm, 'java.sql.DriverManager')
conn = spark._jvm.DriverManager.getConnection(jdbc_url, conn_props["user"], conn_props["password"])
stmt = conn.createStatement()

merge_sql = f"""
MERGE INTO {target_table} AS target
USING {tmp_table} AS source
ON target.col1 = source.col1 AND target.col2 = source.col2
WHEN MATCHED THEN
    UPDATE SET 
        -- 替换为所有需要更新的非主键列
        target.col3 = source.col3,
        target.col4 = source.col4,
        target.datetime_col = source.datetime_col
WHEN NOT MATCHED THEN
    INSERT (col1, col2, col3, col4, datetime_col) -- 替换为目标表所有列
    VALUES (source.col1, source.col2, source.col3, source.col4, source.datetime_col);
"""
stmt.executeUpdate(merge_sql)

# 3. 删除临时表
stmt.executeUpdate(f"DROP TABLE {tmp_table}")

stmt.close()
conn.close()

• SQL Server datetime2的合法取值范围是0001-01-01 00:00:00 ~ 9999-12-31 23:59:59.9999999，需确保你的时间列没有超出该范围的取值，比如1970年之前的时间戳、空值被错误解析为异常时间等。
• 数据量大的时候可以适当调整jdbc的batchsize参数，提升写入性能。

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