Delta Live Table Schema校验未生效，Autoloader处理非法数据未报错求助

我使用Autoloader从存储持续读取数据至Databricks Delta Live Table，数据管道代码如下：

import dlt
from pyspark.sql.functions import *
from pyspark.sql.types import *
    
sch = "StructType([StructField('Date', StringType(), True), StructField('machine', StringType(), True), StructField('temperature', DecimalType(), True), StructField('time', StringType(), True)])" 
    
@dlt.create_table(
  comment="The raw machine data, ingested from azure storage.",
  table_properties={
    "myCompanyPipeline.quality": "raw",
    "pipelines.autoOptimize.managed": "true"
  }
)
def test_raw():
  return (spark.readStream.format("cloudFiles").option("schema",sch).option("cloudFiles.schemaLocation", "/FileStore/schema").option("cloudFiles.format", "json").load("..../"))

读取的数据集包含非法数据：

{"Date":"2023-10-16","time":"12:00:00","machine":"Machine1","temperature":"23.50"}
{"Date":"2023-10-16","time":"12:00:01","machine":"Machine2","temperature":"...corrupt temp..."}
{"Date":"2023-10-16","time":"12:00:02","machine":"Machine3","temperature":"27.50"}

当temperature字段存在非Decimal类型的非法数据时，管道未报错反而成功处理所有记录，按预期应触发失败，为何Schema enforcement未生效？

1. Schema定义与传递错误 #

你将Schema定义成了字符串形式，但cloudFiles.schema选项需要接收的是实际的StructType对象，而非字符串。Autoloader无法解析字符串格式的Schema，会自动退化为模式推断（schema inference），推断出的temperature字段会被识别为StringType，自然不会触发类型校验。

2. 未开启严格的Schema强制校验 #

默认情况下，Autoloader采用宽松的Schema处理逻辑，即使传递了正确的Schema，也需要显式配置校验规则才能让非法数据触发报错。

修正后的代码 #

import dlt
from pyspark.sql.functions import *
from pyspark.sql.types import *

# 定义实际的StructType对象，而非字符串
sch = StructType([
    StructField('Date', StringType(), True),
    StructField('machine', StringType(), True),
    StructField('temperature', DecimalType(), True),
    StructField('time', StringType(), True)
])

@dlt.create_table(
  comment="The raw machine data, ingested from azure storage.",
  table_properties={
    "myCompanyPipeline.quality": "raw",
    "pipelines.autoOptimize.managed": "true"
  }
)
# 结合DLT质量约束，非法数据直接触发管道失败
@dlt.expect_or_fail("valid_temperature", "temperature IS NOT NULL")
def test_raw():
  return (spark.readStream.format("cloudFiles")
          .option("cloudFiles.schema", sch)  # 传递Schema对象
          .option("cloudFiles.schemaLocation", "/FileStore/schema")
          .option("cloudFiles.format", "json")
          # 开启严格Schema校验，字段类型不匹配时标记脏数据
          .option("cloudFiles.schemaValidationMode", "strict")
          # 禁止Schema自动演化，新增字段直接报错
          .option("cloudFiles.schemaEvolutionMode", "failOnNewColumns")
          .load("..../"))

关键配置说明 #

• cloudFiles.schemaValidationMode = strict：严格校验字段类型与定义的Schema匹配，不匹配的记录会被标记为脏数据。
• @dlt.expect_or_fail("valid_temperature", "temperature IS NOT NULL")：DLT的质量约束，当存在temperature为NULL的脏数据时，直接让管道运行失败，符合预期的触发失败逻辑。

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