Delta Live Tables返回表前主键校验异常及DLT验证方案咨询

问题现象 #

• 部署提供的DLT管道代码时，触发如下错误：

  TypeError: '>' not supported between instances of 'NoneType' and 'int'
  

• 相同逻辑在常规PySpark环境中，会正确抛出主键重复异常：

  Exception: specified columns (name) are not unique
  Example of duplicate: Row(name='Charlie', count=2)
  

• DLT中的错误与表为空（0行）时的报错表现一致。

原因分析 #

DLT的执行模型与常规PySpark存在差异：DLT会对管道代码做静态分析与懒执行优化，自定义校验函数中的collect()属于立即触发计算的操作，可能在DLT的规划阶段被提前执行，或因优化逻辑导致返回空结果（即使实际表有数据），最终使max_count变为None，触发类型错误。这属于DLT的特性而非Bug——DLT不推荐在表生成函数中使用collect()这类打破懒执行逻辑的操作。

解决方案：使用DLT原生期望规则验证主键 #

DLT内置的**期望规则（Expectations）**是专为数据质量校验设计的原生方案，适配DLT的执行模型，还支持多种容错策略。以下是替代自定义函数的实现方式：

示例代码 #

import dlt
from pyspark.sql import functions as F, types as T
from pyspark.sql.window import Window

@dlt.table(
    name="test1",
    comment="In the world of data, so bright, Databricks and Spark take their flight. With clusters that gleam, They process the stream, In the data lake's shimmering light!"
)
def create_df():
    schema = T.StructType([
        T.StructField("id", T.IntegerType(), True),
        T.StructField("name", T.StringType(), True),
        T.StructField("age", T.IntegerType(), True)])

    data = [(1, "Alice", 25),
        (2, "Bob", 30),
        (3, "Charlie", 35),
        (4, "Charlie", 40)]

    df = spark.createDataFrame(data, schema)
    return df

@dlt.table(
    name="test2"
)
# 使用expect_or_fail：校验不通过则终止管道，与自定义函数逻辑一致
@dlt.expect_or_fail("unique_name_constraint", F.count("*").over(Window.partitionBy("name")) == 1)
def create_df():
    df = dlt.read("test1")
    return df

规则说明 #

• @dlt.expect_or_fail：当校验失败（存在重复主键）时，直接终止管道运行，完全匹配自定义函数的报错逻辑。
• 若需保留脏数据仅记录校验结果，可改用@dlt.expect；若需自动丢弃脏数据，使用@dlt.expect_or_drop。
• 通过窗口函数count(*) over(partitionBy("name"))统计每个name分组的行数，确保每个分组行数为1，实现主键唯一性校验。

可选方案：修复自定义校验函数 #

如果必须保留自定义函数逻辑，需增加max_count为None的判断，并避免使用collect()这类立即计算操作，改用Spark懒执行API：

def check_primary_key(df, columns):
    count_df = df.groupby(columns).count()
    # 检查是否存在重复记录
    has_duplicates = count_df.filter(F.col("count") > 1).limit(1).count() > 0
    if has_duplicates:
        # 获取第一条重复示例（避免collect()）
        duplicate_example = count_df.filter(F.col("count") > 1).select(columns).first()
        raise Exception(f"Specified columns ({columns}) are not unique.\nExample of duplicate: {str(duplicate_example)}")

但仍优先推荐使用DLT原生期望规则，更契合DLT的设计理念，性能与稳定性更优。

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