从SQL Server向Delta Lake加载数据时的数据类型不匹配问题求助

方案1：动态对齐源表列类型到目标表Schema #

利用参数化Notebook的特性，自动读取目标Delta表的Schema，将源DataFrame的列类型批量转换为目标表对应列的类型，从根源避免类型冲突。

代码示例：

# 从参数中获取目标Delta表路径
target_delta_path = "s3 path"

# 读取目标Delta表的Schema
target_schema = spark.read.format("delta").load(target_delta_path).schema

# 动态转换列类型
def align_column_type(df, col_name, target_type):
    current_type = df.schema[col_name].dataType
    # 处理Long与Decimal的双向兼容转换（可根据业务调整规则）
    if isinstance(current_type, LongType) and isinstance(target_type, DecimalType):
        return df.withColumn(col_name, col(col_name).cast(target_type))
    elif isinstance(current_type, DecimalType) and isinstance(target_type, LongType):
        # 注意：Decimal转Long需确保数据在Long范围内，避免精度丢失
        return df.withColumn(col_name, col(col_name).cast(target_type))
    else:
        return df.withColumn(col_name, col(col_name).cast(target_type))

# 遍历所有列完成类型对齐
for field in target_schema.fields:
    if field.name in DF.columns:
        DF = align_column_type(DF, field.name, field.dataType)

# 执行写入，保留mergeSchema处理新增列等兼容变更
DF.write.mode("overwrite").format("delta").option("mergeSchema", "true").save(target_delta_path)

方案2：前置Schema安全校验与拦截 #

在写入前添加校验逻辑，仅允许预设的兼容类型转换，遇到不兼容变更直接抛出异常，避免意外修改目标表Schema。

代码示例：

from pyspark.sql.types import LongType, DecimalType, StringType

# 定义允许的类型转换规则（源类型→允许的目标类型列表）
allowed_conversions = {
    LongType: [DecimalType, LongType],
    DecimalType: [LongType, DecimalType],
    StringType: [StringType]
    # 根据业务需求扩展其他类型规则
}

target_schema = spark.read.format("delta").load(target_delta_path).schema
conflict_list = []

# 遍历列对比类型兼容性
for field in target_schema.fields:
    if field.name in DF.columns:
        source_type = DF.schema[field.name].dataType
        target_type = field.dataType
        if type(source_type) not in allowed_conversions or type(target_type) not in allowed_conversions[type(source_type)]:
            conflict_list.append(f"列{field.name}类型不兼容：源[{source_type}] → 目标[{target_type}]")

# 校验不通过则终止流程
if conflict_list:
    raise Exception("Schema校验失败：" + "; ".join(conflict_list))
else:
    DF.write.mode("overwrite").format("delta").option("mergeSchema", "true").save(target_delta_path)

方案3：结合Delta Lake Schema Evolution的灰度校验 #

如果需要保留mergeSchema的自动扩展能力，可先将源数据写入临时Delta表，对比临时表与目标表的Schema差异，确认仅存在兼容类型变更后，再将临时表数据合并到目标表。这种方式既能处理兼容变更，又能拦截不兼容的类型冲突。

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