📚 课程目录第4章数据质量规则引擎

4. 数据质量规则引擎：规则定义与分类

规则引擎这东西，说白了就是质量平台的「大脑」。我见过不少团队，一上来就堆规则，结果跑起来全是误报，运维同学天天骂街。其实核心就三件事：规则怎么定义、怎么管、怎么跑。咱们一个一个说。

4.1 规则分类：四种最常见的检查类型

我在项目中习惯把规则分成四大类。你想想看，数据质量问题翻来覆去也就那么几种：缺了、重复了、超范围了、格式乱了。

规则类型	检查目标	典型场景	常见误报原因
空值检查	字段是否为 NULL 或空串	用户手机号、订单金额	业务上允许为空的历史数据
唯一性检查	字段或组合值是否重复	身份证号、订单号	多源合并时的时间戳差异
范围检查	数值/日期是否在合理区间	年龄 0-120、下单时间不晚于今天	跨时区导致的日期偏移
格式检查	是否符合正则或枚举值	邮箱格式、性别枚举	国际化场景下的编码差异

我曾经踩过一个坑： 空值检查直接用了 IS NULL，结果漏掉了空串 ''。后来统一改成 COALESCE(TRIM(col), '') = ''，才把问题堵住。

4.2 规则配置与管理：别让规则变成「死规则」

规则配好了，怎么管？我建议用三层结构：规则模板 → 规则实例 → 规则运行参数。这样能避免重复造轮子。

4.2.1 规则模板定义

模板就是规则的「骨架」。比如空值检查模板，只需要定义检查字段和阈值。我个人习惯用 JSON Schema 来描述模板：

{
  "ruleType": "NOT_NULL",
  "templateId": "TMPL_001",
  "params": {
    "targetColumn": "string",
    "nullThreshold": "number",
    "scope": "string"  // ALL | RECENT_7D | CUSTOM
  },
  "output": {
    "failCount": "integer",
    "failRate": "float",
    "sampleRecords": "array"
  }
}

4.2.2 规则实例化

模板配好后，针对具体表创建实例。比如对 ods_order 表的 phone 字段创建一个空值检查实例：

{
  "instanceId": "INST_20241001_001",
  "templateId": "TMPL_001",
  "targetTable": "ods_order",
  "targetColumn": "phone",
  "params": {
    "nullThreshold": 0.01,
    "scope": "RECENT_7D"
  },
  "schedule": "0 0 6 * * ?",  // 每天早上6点
  "enabled": true
}

一个小技巧： 阈值别设死。我一般用「动态阈值」——基于过去7天的统计值自动计算，比如 avg_fail_rate * 1.5。这样能适应业务波动，减少误报。

4.3 规则执行引擎设计

引擎设计是重头戏。我参与过两版引擎重构，第一版是纯单机跑，第二版改成了分层调度 + 并行执行。核心架构我画了个图，你一看就明白：

4.3.1 调度策略

调度器我推荐用两级调度：

全局调度：每天凌晨跑全量规则，按表分区并行
增量调度：实时或准实时（每5分钟）跑增量数据上的规则

为什么要分开？因为全量跑一次可能要几小时，增量跑必须秒级响应。混在一起，调度器会卡死。

4.3.2 执行器设计要点

执行器这块，我踩过最大的坑是资源隔离。曾经有个大表跑唯一性检查，直接把数据库连接池打满了，导致其他规则全部超时。后来我做了三件事：

线程池隔离：大表规则用独立线程池，最多4个并发
超时熔断：单条规则超过30分钟自动终止，记录超时原因
结果采样：失败记录超过10万条时，只保留前1000条样本，避免内存溢出

核心原则： 规则引擎不能成为数据库的「杀手」。所有检查SQL必须走只读副本，并且强制加 LIMIT 和超时时间。

4.3.3 执行流程伪代码

我习惯用伪代码来描述流程，比文字更直观：

function executeRules(rules, tablePartitions):
    // 1. 依赖排序
    sortedRules = topologicalSort(rules)
    
    // 2. 分组执行
    for group in sortedRules:
        parallelSubmit(group, executorPool)
    
    // 3. 收集结果
    results = []
    for future in futures:
        try:
            result = future.get(timeout=1800)
            results.append(result)
        except TimeoutException:
            results.append({
                "ruleId": future.ruleId,
                "status": "TIMEOUT",
                "message": "执行超过30分钟"
            })
    
    // 4. 聚合报告
    return aggregateReport(results)

我个人的习惯： 每个规则实例都带上 run_id 和 snapshot_time。这样出问题时，可以精确回溯到「某次运行」的数据快照，而不是查当前表。这个习惯救过我很多次。

4.4 规则生命周期管理

规则不是配完就不管了。我建议给规则加上状态机：

状态	含义	触发条件
草稿	规则刚创建，未启用	新建规则
启用	正常调度执行	手动启用 / 审批通过
暂停	暂时停止执行	手动暂停 / 连续误报超过阈值
废弃	不再使用，保留历史	手动废弃 / 表下线自动触发

嗯，这里要注意：废弃的规则不要物理删除。我见过有人直接 DELETE 掉，结果审计的时候查不到历史记录，被罚了整改。软删除加个 status 字段就够了。

好了，规则引擎的核心设计思路就这些。从规则分类到配置管理，再到执行引擎的架构，每一步都有坑，但每一步也都有成熟的解法。你按这个思路搭，至少能保证跑得稳、查得清、改得动。