Databricks Delta表decimal(10,0)转decimal(38,18)修改列失败问题

根因梳理 #

• 首次建表类型降级：你指定的decimal(38,18)被Spark默认配置或Delta旧版本协议限制，自动降级为DecimalType(10,0)，大概率是未开启Delta列映射、集群默认decimal精度配置为10/0、写入空DataFrame时未强制保留类型三个原因共同导致。
• 流写入无输出无日志：foreachBatch中的MERGE操作遇到列类型不匹配时错误被结构化流容错机制吞掉，无显式抛出，且类型不匹配的行会被直接丢弃不会写入。
• ALTER TABLE报错：默认Delta表协议不支持decimal精度扩容操作，需要先升级表协议、开启列映射后才能执行类型修改。

分步解决方案 #

方案一：测试表无存量数据，直接重建（最快） #

如果表无业务数据可直接重建，能彻底规避类型兼容问题：

1. 清理原表和存储路径

DROP TABLE IF EXISTS mytable;

对应删除表关联的Delta存储路径下的所有文件。
2. 写入空DataFrame时强制保留类型配置

from pyspark.sql.types import StructType, DecimalType
# 此处读取你的JSON schema生成StructType，确保income字段显式定义为DecimalType(38,18)
your_struct_type = 你的StructType生成逻辑
empty_df = spark.createDataFrame([], schema=your_struct_type)
# 写入时加Delta配置强制保留高精度类型
empty_df.write \
  .format("delta") \
  .option("delta.columnMapping.mode", "name") \
  .option("delta.minWriterVersion", "5") \
  .option("delta.minReaderVersion", "2") \
  .mode("overwrite") \
  .save("你的Delta表存储路径")

3. 重新建表关联路径

CREATE TABLE mytable
USING DELTA
LOCATION '你的Delta表存储路径';

执行DESCRIBE mytable;验证income字段类型为decimal(38,18)即可。

方案二：生产表有存量数据，原地修改类型 #

如果表有业务数据不能删除，先升级表协议再修改字段类型：

1. 先升级Delta表协议，开启列映射

ALTER TABLE mytable SET TBLPROPERTIES (
  'delta.minReaderVersion' = '2',
  'delta.minWriterVersion' = '5',
  'delta.columnMapping.mode' = 'name'
);

2. 执行字段类型修改

ALTER TABLE mytable ALTER COLUMN income TYPE DECIMAL(38,18);

注：该操作仅支持Databricks Runtime 10.4 LTS及以上版本，属于向上兼容修改不会丢失存量数据

方案三：修复结构化流写入逻辑，加错误日志 #

修改完表字段类型后，在foreachBatch中添加异常捕获逻辑，避免后续问题无报错：

def upsert_batch(batch_df, batch_id):
    try:
        batch_df.createOrReplaceTempView("stream_batch")
        # 替换为你的实际MERGE逻辑
        spark.sql("""
            MERGE INTO mytable t
            USING stream_batch s
            ON t.unique_id = s.unique_id
            WHEN MATCHED THEN UPDATE SET *
            WHEN NOT MATCHED THEN INSERT *
        """)
    except Exception as e:
        print(f"批次{batch_id}写入失败，错误：{str(e)}")
        # 按需抛出异常触发作业告警
        raise e

# 结构化流写入配置
stream_df.writeStream \
  .foreachBatch(upsert_batch) \
  .option("checkpointLocation", "你的流Checkpoint存储路径") \
  .trigger(availableNow=True) \
  .start()

后续规避方案 #

• 所有Decimal字段在StructType中显式指定精度和小数位，不要使用无参数的DecimalType()
• 集群Spark配置添加两个参数：spark.sql.decimalType.defaultPrecision=38、spark.sql.decimalType.defaultScale=18，避免默认降级decimal类型
• 新建Delta表时统一开启列映射配置，避免类型、列名相关的兼容问题

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