PySpark JDBC插入SQL Server时，如何正确处理None/numpy.nan值？

我通过PySpark DataFrame执行追加操作向SQL Server已有表插入数据，使用的命令如下：

df_spark.write.jdbc(url=jdbcUrl, table='scd.'+table_name, mode='append' ,properties=connectionProperties)

目标表结构为：

CREATE TABLE [scd].[test]([Id] INT,
                          [InsertDatetime] datetime2 NULL)

该表中InsertDatetime列是允许为NULL的datetime2类型，但尝试插入空值时遇到以下问题：

1. 使用Python None作为空值 #

代码示例：

a = {
    'Id':[1, 2],
    'InsertDatetime':['2022-11-24T09:56:25.7228705', None]
}
a = pd.DataFrame(a)
df_spark = spark.createDataFrame(a)
table_name = 'test'
df_spark.write.jdbc(url=jdbcUrl, table='scd.'+table_name, mode='append', properties=connectionProperties)

报错信息：

java.lang.IllegalArgumentException: Can't get JDBC type for void

2. 使用numpy.nan作为空值 #

代码示例：

a = {
    'Id':[1, 2],
    'InsertDatetime':['2022-11-24T09:56:25.7228705', np.nan]
}
a = pd.DataFrame(a)
df_spark = spark.createDataFrame(a)
table_name = 'test'
df_spark.write.jdbc(url=jdbcUrl, table='scd.'+table_name, mode='append' ,properties=connectionProperties)

报错信息：

java.sql.BatchUpdateException: Operand type clash: float is incompatible with datetime2

解决方案 #

方法一：先将Pandas DataFrame的日期列转为datetime类型 #

Pandas中混合字符串和None会导致列类型为object，Spark无法正确识别为Timestamp类型；而用np.nan则会让列变为float类型，和SQL Server的datetime2冲突。正确做法是先把日期列转为datetime类型，此时None会被转为pd.NaT，Spark能正确识别为Timestamp类型的NULL：

import pandas as pd

a = {
    'Id':[1, 2],
    'InsertDatetime':['2022-11-24T09:56:25.7228705', None]
}
df_pandas = pd.DataFrame(a)
# 转换日期列为datetime类型，自动将None转为NaT
df_pandas['InsertDatetime'] = pd.to_datetime(df_pandas['InsertDatetime'])

df_spark = spark.createDataFrame(df_pandas)
table_name = 'test'
df_spark.write.jdbc(url=jdbcUrl, table='scd.'+table_name, mode='append', properties=connectionProperties)

方法二：直接用Spark指定Schema创建DataFrame #

跳过Pandas，直接用Spark创建DataFrame并指定Timestamp类型，确保空值被正确识别：

from pyspark.sql import types as T

# 定义Schema，明确指定InsertDatetime为Timestamp类型且允许为空
schema = T.StructType([
    T.StructField("Id", T.IntegerType(), nullable=False),
    T.StructField("InsertDatetime", T.TimestampType(), nullable=True)
])

# 构造数据，None直接作为空值
data = [
    (1, "2022-11-24T09:56:25.7228705"),
    (2, None)
]

df_spark = spark.createDataFrame(data, schema=schema)
df_spark.write.jdbc(url=jdbcUrl, table='scd.test', mode='append', properties=connectionProperties)

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