数据契约的核心三要素

数据契约这东西，说白了就是一份「数据承诺书」。我做了这么多年数据架构，见过太多因为数据质量翻车的案例。后来发现，只要把三个核心要素定清楚，大部分问题都能提前规避。

哪三个？Schema定义、约束规则、数据质量指标。咱们一个一个拆开聊。

Schema定义：数据的骨架

Schema定义是数据契约的基石。它规定了数据长什么样——有哪些字段、每个字段是什么类型、字段之间怎么组织。

在Delta Lake里，Schema定义天然就是强类型的。你写Spark DataFrame时定义的StructType，就是最直接的Schema契约。

举个例子，一个用户信息表的Schema定义：

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

user_schema = StructType([
    StructField("user_id", StringType(), nullable=False),
    StructField("user_name", StringType(), nullable=False),
    StructField("age", IntegerType(), nullable=True),
    StructField("email", StringType(), nullable=False),
    StructField("created_at", TimestampType(), nullable=False),
    StructField("updated_at", TimestampType(), nullable=True)
])

我个人习惯，每个字段都会加一段中文描述，写在字段注释里。这样后来接手的人一看就明白，不用翻文档。

约束规则：数据的红线

Schema定义只是第一步。真正让数据可信的，是约束规则。我把它分成三类：非空约束、唯一约束、枚举约束。

非空约束

这个最简单，但也是最容易被忽略的。你想想看，一个用户表里user_id是空的，这数据还能用吗？

在Delta Lake里，非空约束可以直接在Schema里声明：

StructField("user_id", StringType(), nullable=False)

写入数据时，如果某条记录的user_id是null，Delta Lake会直接报错。嗯，这里要注意——这个报错是写入时发生的，不是事后检查。所以你的ETL流程必须处理这个异常。

唯一约束

唯一约束保证某个字段或字段组合的值在整个表里不重复。这在主键场景下特别重要。

Delta Lake原生不支持唯一约束，但我们可以用MERGE操作来模拟：

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/path/to/user_table")

delta_table.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.user_id = source.user_id"
    ) \
    .whenNotMatchedInsertAll() \
    .execute()

这段代码的逻辑是：如果user_id已存在，就不插入；只有新user_id才写入。说白了，这就是在写入时做唯一性校验。

枚举约束

枚举约束限制字段的值只能从指定集合里选。比如订单状态字段，只能是"待支付"、"已支付"、"已发货"、"已完成"、"已取消"这五个值。

在Delta Lake里，可以用CHECK约束来实现：

ALTER TABLE user_table ADD CONSTRAINT status_check 
CHECK (status IN ('待支付', '已支付', '已发货', '已完成', '已取消'))

写入数据时，如果status字段的值不在这个列表里，Delta Lake会拒绝写入。

数据质量指标：数据的体检报告

约束规则是硬性的——要么通过，要么拒绝。但有些场景下，数据质量是「程度」问题，不是「对错」问题。这时候就需要数据质量指标。

我常用的三个指标：完整性、准确性、一致性。

指标	定义	Delta Lake中的实现方式
完整性	数据是否完整，有没有缺失	非空约束 + 行数校验
准确性	数据是否真实反映业务	CHECK约束 + 业务规则校验
一致性	数据在不同系统间是否一致	跨表JOIN校验 + 时间戳比对

完整性

完整性不只是非空。它还包含：记录数是否在预期范围内、关键字段的覆盖率是否达标。

举个例子，我做过一个日活用户统计的管道。数据契约里规定：每天至少要有100万条记录，少于这个数就报警。这个校验我用Delta Lake的DESCRIBE HISTORY来获取写入行数，然后跟阈值比较。

history_df = delta_table.history()
last_commit = history_df.orderBy("version", ascending=False).first()
row_count = last_commit["operationMetrics"]["numOutputRows"]

if int(row_count) < 1000000:
    raise ValueError(f"数据完整性校验失败：预期100万行，实际{row_count}行")

准确性

准确性校验最头疼。因为它依赖业务规则，而业务规则经常变。

我的做法是：把准确性校验规则也写进数据契约，用Delta Lake的CHECK约束来兜底。比如金额字段不能为负数：

ALTER TABLE transaction_table ADD CONSTRAINT amount_positive 
CHECK (amount >= 0)

但有些准确性规则没法用CHECK约束表达。比如「订单金额 = 商品单价 × 数量」。这种我就在ETL脚本里加一个校验步骤，校验不通过就写入一个专门的异常表。

一致性

一致性校验，说白了就是看不同表、不同系统之间的数据对不对得上。

在Delta Lake里，我常用的是跨表JOIN校验。比如用户表和订单表，要保证每个订单的user_id在用户表里都存在：

invalid_orders = spark.sql("""
    SELECT o.order_id, o.user_id
    FROM orders o
    LEFT JOIN users u ON o.user_id = u.user_id
    WHERE u.user_id IS NULL
""")

if invalid_orders.count() > 0:
    # 写入异常表，触发告警
    invalid_orders.write.format("delta").mode("append") \
        .save("/path/to/data_quality_issues")

这里要注意，一致性校验通常是事后做的，不是写入时做。因为写入时做JOIN校验，性能开销太大。

核心要素的关系

这三个要素不是孤立的。它们层层递进：

• Schema定义是基础，决定了数据长什么样
• 约束规则是防线，阻止坏数据进入
• 数据质量指标是监控，持续评估数据健康度

下面这张图展示了它们的关系：

从这张图能看出来，数据契约不是一锤子买卖。它是从定义到约束再到监控的完整闭环。我个人习惯，每张表的数据契约都包含这三层，缺一层都不放心。

写在Delta Lake里的实践建议

最后，分享几个我在项目中踩过的坑和总结的经验：

• Schema演化要谨慎：Delta Lake支持Schema演化，但数据契约里的Schema定义不能随便改。我建议用saveAsTable时指定mergeSchema=False，强制Schema匹配。
• 约束规则要分层：硬约束（非空、枚举）放在写入阶段，软约束（唯一性、准确性）放在事后校验。这样既保证数据可用性，又不会阻塞写入。
• 质量指标要可量化：不要只说「数据要完整」，要说「完整率不低于99.9%」。可量化的指标才能自动化监控。
• 异常数据要留痕：校验不通过的数据不要直接丢弃，写入异常表。这样事后排查问题时，有据可查。