Databricks连接SQL Server视图时日期转换错误及优化方案求助

问题背景 #

使用Databricks的SQL DW连接器读取SQL Server视图，代码如下：

df = spark.read
          .format("com.databricks.spark.sqldw")
          .option("url", url)
          .option("dbtable", table_name)
          .option("user", key)
          .option("password", password)
          .load()

视图中部分日期值以varchar类型存储，格式统一为yyyy-MM-dd，无空值或损坏数据。执行聚合操作时触发错误：

plan_df.groupBy("FrozenDateKey")
        .agg(
            F.min(planned_date).alias("min_planned_date"),
            F.max(planned_date).alias("max_planned_date"),
        )

错误信息：

Conversion failed when converting date and/or time from character string

转为RDD再转回DataFrame可规避错误，但资源消耗大、效率低。

优化方案 #

1. 读取时显式指定Schema，强制日期列类型

Spark自动推断Schema时可能误判varchar类型的日期列为字符串，导致聚合计算被下推到SQL Server，引发转换错误。显式定义Schema可让Spark在读取阶段就完成类型转换：

from pyspark.sql.types import StructType, StructField, DateType, StringType, IntegerType

# 根据视图实际结构定义Schema，将日期列设为DateType
custom_schema = StructType([
    StructField("FrozenDateKey", IntegerType(), True),
    StructField("planned_date", DateType(), True),
    # 其他列按实际类型补充定义
])

df = spark.read
          .format("com.databricks.spark.sqldw")
          .schema(custom_schema)
          .option("url", url)
          .option("dbtable", table_name)
          .option("user", key)
          .option("password", password)
          .load()

2. 读取后立即转换日期列类型

如果无法提前确定完整Schema，可在读取后用to_date函数显式转换字符串列为日期类型，后续聚合使用转换后的列：

from pyspark.sql import functions as F

# 按指定格式将字符串列转为日期类型
plan_df = df.withColumn("planned_date", F.to_date(F.col("planned_date"), "yyyy-MM-dd"))

# 基于转换后的日期列执行聚合
plan_df.groupBy("FrozenDateKey")
        .agg(
            F.min("planned_date").alias("min_planned_date"),
            F.max("planned_date").alias("max_planned_date"),
        )

3. 禁用聚合下推，让计算在Spark端执行

错误核心是Spark将聚合操作下推到SQL Server，而SQL Server无法正确转换varchar到日期类型。禁用聚合下推后，Spark会在本地完成类型转换与聚合计算：

df = spark.read
          .format("com.databricks.spark.sqldw")
          .option("url", url)
          .option("dbtable", table_name)
          .option("user", key)
          .option("password", password)
          .option("pushDownAggregate", "false")  # 禁用聚合下推
          .load()

关于直接加载为RDD

不建议直接加载为RDD，RDD API的类型安全性和执行效率远低于DataFrame。转RDD再转回DF的本质是触发数据全量加载到Spark内存后重新解析类型，资源消耗远高于上述DataFrame层面的优化方案。

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