3. 数据校验工具与框架：Great Expectations、Deequ、Apache Griffin、dbt test、自定义Python脚本

聊到数据校验工具，我见过不少团队一开始就想着自己写脚本。说实话，自己写确实灵活，但维护成本高得吓人。今天我把市面上主流的几个工具掰开揉碎讲一讲，包括我自己踩过的坑。

3.1 Great Expectations：社区最活跃的校验框架

Great Expectations（简称GE）是我个人最常用的工具之一。它把数据校验抽象成「Expectation」这个概念，说白了就是你对数据应该长什么样的预期。

举个例子，你想校验订单表的金额字段不能为负：

import great_expectations as ge

# 加载数据
df = ge.read_csv("orders.csv")

# 定义预期：amount字段必须大于等于0
df.expect_column_values_to_be_between(
    column="amount",
    min_value=0,
    max_value=1000000
)

# 执行校验并生成报告
results = df.validate()
print(results)

我在项目中遇到过一个问题：GE默认会把所有数据加载到内存里。如果你的表有上亿行，直接崩。后来我改用Spark Execution Engine才解决。

3.2 Deequ：亚马逊开源的Spark原生校验库

Deequ是亚马逊基于Spark开发的校验库。它最大的优势是——跑在大数据集群上，不占本地内存。

特性	Great Expectations	Deequ
运行引擎	Pandas / Spark	Spark only
数据规模	中小规模（内存限制）	大规模（分布式）
规则定义	Python DSL	Scala/Python DSL
报告生成	内置Data Docs	需自定义
社区活跃度	高	中

Deequ的校验规则用起来也很直观：

import com.amazon.deequ.VerificationSuite
import com.amazon.deequ.checks.{Check, CheckLevel}

val verificationResult = VerificationSuite()
  .onData(spark.table("orders"))
  .addCheck(
    Check(CheckLevel.Error, "订单数据完整性校验")
      .isComplete("order_id")          // 不为空
      .isUnique("order_id")            // 唯一
      .isNonNegative("amount")         // 非负
      .isContainedIn("status", Array("pending", "paid", "cancelled"))
  )
  .run()

嗯，这里要注意：Deequ的Python API不如Scala成熟。我建议如果团队以Scala为主，优先选Deequ；如果是纯Python栈，还是GE更顺手。

3.3 Apache Griffin：企业级数据质量平台

Apache Griffin是一个完整的数据质量平台，不是单纯的校验库。它提供了Web UI、调度引擎、告警机制，说白了就是「开箱即用」的企业级方案。

Griffin的架构分为三层：

Griffin的优点是全，缺点是重。我记得有一次帮客户部署Griffin，光环境配置就花了两天。如果你的团队只有两三个人维护数仓，我建议别碰Griffin，太沉了。

3.4 dbt test：数仓工程师的瑞士军刀

dbt test是dbt生态的一部分。如果你已经在用dbt做数据建模，那dbt test就是最自然的校验方案——不需要额外部署，不需要学习新框架。

dbt test分两种：

• 内置测试：not_null, unique, accepted_values, relationships
• 自定义测试：用SQL写任意校验逻辑

看个例子，在schema.yml里定义测试：

version: 2

models:
  - name: orders
    columns:
      - name: order_id
        tests:
          - not_null
          - unique
      - name: amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1000000
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'paid', 'cancelled']

跑测试就一行命令：

dbt test --select orders

我个人特别喜欢dbt test的一点是——它和dbt的文档生成无缝集成。跑完测试，dbt docs serve里直接能看到哪些字段通过了校验，哪些挂了。你想想看，这比翻日志文件舒服多了。

3.5 自定义Python脚本：最后的底牌

说了这么多工具，什么时候该自己写脚本？我总结三个场景：

1. 校验逻辑极其特殊：比如要结合外部API判断数据是否合法
2. 数据源不在工具支持列表里：比如某些自研的存储引擎
3. 性能要求极致：工具框架有额外开销，裸写SQL或PySpark更快

我自己写校验脚本的模板长这样：

import pandas as pd
import logging

logger = logging.getLogger(__name__)

def validate_orders(df: pd.DataFrame) -> dict:
    """
    自定义订单数据校验
    返回：{ "passed": bool, "errors": list, "warnings": list }
    """
    result = {"passed": True, "errors": [], "warnings": []}
    
    # 1. 完整性校验
    null_count = df["order_id"].isnull().sum()
    if null_count > 0:
        result["passed"] = False
        result["errors"].append(f"order_id 存在 {null_count} 条空值")
    
    # 2. 业务规则校验
    negative_amount = df[df["amount"] < 0]
    if len(negative_amount) > 0:
        result["passed"] = False
        result["errors"].append(f"存在 {len(negative_amount)} 条负金额记录")
    
    # 3. 波动性校验（与历史对比）
    avg_amount = df["amount"].mean()
    if avg_amount > 1000:  # 假设历史均值在500左右
        result["warnings"].append(f"平均金额 {avg_amount:.2f} 异常偏高")
    
    return result

# 使用
df = pd.read_parquet("/data/orders/dt=2024-01-01/")
check_result = validate_orders(df)
if not check_result["passed"]:
    logger.error(f"校验失败: {check_result['errors']}")
    # 触发告警或阻断下游

自己写脚本最大的坑是什么？我曾经吃过一次大亏——脚本里忘了处理空指针异常，结果校验作业跑了两个小时，最后抛了个NullPointerException，所有数据都白跑了。所以我的建议是：能用现成工具就别自己写，除非你愿意花时间把异常处理、日志、告警、重试机制都写好。

3.6 工具选型决策矩阵

最后给一张决策表，帮你快速选型：

场景	推荐工具	理由
小团队、Python栈、快速上手	Great Expectations	社区活跃、文档完善、集成CI方便
大数据量、Spark生态	Deequ	分布式执行、不占本地内存
企业级平台、需要UI和告警	Apache Griffin	全链路管理、开箱即用
已用dbt做建模	dbt test	零额外成本、与建模无缝集成
特殊逻辑或小众数据源	自定义Python脚本	灵活、无依赖

说白了，没有银弹。我见过用GE跑百亿级数据跑崩了的，也见过用dbt test做复杂跨表校验做不了的。关键是根据你的数据规模、技术栈、团队能力来选。如果拿不准，我建议从小处着手——先选一个工具跑通一条链路，再逐步推广。