数据质量规则引擎设计实战：从规则配置到DQI评分的全链路实现方案

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

rule_id: "RULE_001"
rule_name: "用户表主键非空校验"
rule_type: "完整性"
table_name: "dwd.dwd_user_info_di"
check_column: "user_id"
check_logic: "user_id is not null"
filter_condition: "dt = '@dt'"
execute_frequency: "daily"
alarm_threshold: 0.01 # 异常率超过1%则告警
alarm_channels: ["dingtalk", "email"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

def calculate_dqi(quality_metrics: dict) -> tuple[float, str]:
    """
    计算DQI评分
    :param quality_metrics: 各维度质量指标，包含缺失率、重复率、错误率、不一致率、是否按时产出
    :return: DQI分数，质量等级
    """
    # 各维度权重
    weights = {
        "completeness": 0.3,
        "uniqueness": 0.2,
        "accuracy": 0.3,
        "consistency": 0.15,
        "timeliness": 0.05
    }
    
    # 计算各维度得分
    completeness_score = (1 - quality_metrics["missing_rate"]) * 100 * weights["completeness"]
    uniqueness_score = (1 - quality_metrics["duplicate_rate"]) * 100 * weights["uniqueness"]
    accuracy_score = (1 - quality_metrics["error_rate"]) * 100 * weights["accuracy"]
    consistency_score = (1 - quality_metrics["inconsistent_rate"]) * 100 * weights["consistency"]
    timeliness_score = 100 * weights["timeliness"] if quality_metrics["is_on_time"] else 0
    
    dqi_score = completeness_score + uniqueness_score + accuracy_score + consistency_score + timeliness_score
    
    # 确定质量等级
    if dqi_score >= 90:
        level = "优秀"
    elif dqi_score >= 80:
        level = "良好"
    elif dqi_score >= 60:
        level = "合格"
    else:
        level = "不合格"
    
    return round(dqi_score, 2), level

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

from pyspark.sql import SparkSession
import yaml

class RuleExecutor:
    def __init__(self, spark: SparkSession):
        self.spark = spark
        self.rule_template_map = self._load_rule_templates()
    
    def _load_rule_templates(self) -> dict:
        """加载规则模板库"""
        with open("rule_templates.yaml", "r") as f:
            return yaml.safe_load(f)
    
    def execute_rule(self, rule_config: dict, dt: str) -> dict:
        """
        执行单个质量规则
        :param rule_config: 规则配置
        :param dt: 数据日期
        :return: 规则执行结果
        """
        # 替换模板变量
        check_sql = self.rule_template_map[rule_config["rule_type"]].format(
            table_name=rule_config["table_name"],
            check_column=rule_config["check_column"],
            check_logic=rule_config["check_logic"],
            dt=dt
        )
        
        # 执行校验SQL
        df = self.spark.sql(check_sql)
        result = df.collect()[0]
        
        total_count = result.total_count
        abnormal_count = result.abnormal_count
        abnormal_rate = abnormal_count / total_count if total_count > 0 else 0
        
        # 采样异常数据
        abnormal_sample = []
        if abnormal_count > 0:
            sample_sql = f"SELECT * FROM {rule_config['table_name']} WHERE dt = '{dt}' AND NOT ({rule_config['check_logic']}) LIMIT 10"
            sample_df = self.spark.sql(sample_sql)
            abnormal_sample = [row.asDict() for row in sample_df.collect()]
        
        return {
            "rule_id": rule_config["rule_id"],
            "rule_name": rule_config["rule_name"],
            "total_count": total_count,
            "abnormal_count": abnormal_count,
            "abnormal_rate": round(abnormal_rate, 4),
            "abnormal_sample": abnormal_sample,
            "is_alarm": abnormal_rate > rule_config["alarm_threshold"]
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

完整性: >
    SELECT 
        COUNT(*) AS total_count,
        SUM(CASE WHEN NOT ({check_logic}) THEN 1 ELSE 0 END) AS abnormal_count
    FROM {table_name}
    WHERE dt = '{dt}'
唯一性: >
    SELECT 
        COUNT(*) AS total_count,
        COUNT(*) - COUNT(DISTINCT {check_column}) AS abnormal_count
    FROM {table_name}
    WHERE dt = '{dt}'
准确性: >
    SELECT 
        COUNT(*) AS total_count,
        SUM(CASE WHEN NOT ({check_logic}) THEN 1 ELSE 0 END) AS abnormal_count
    FROM {table_name}
    WHERE dt = '{dt}'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

from datetime import datetime, timedelta

def daily_dqi_calculation_task(spark: SparkSession):
    """每日DQI计算调度任务"""
    dt = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
    
    # 加载所有生效规则
    with open("active_rules.yaml", "r") as f:
        active_rules = yaml.safe_load(f)
    
    executor = RuleExecutor(spark)
    rule_results = []
    
    # 执行所有规则
    for rule in active_rules:
        result = executor.execute_rule(rule, dt)
        rule_results.append(result)
    
    # 按表聚合计算DQI
    table_metrics = {}
    for result in rule_results:
        table_name = active_rules[[r["rule_id"] == result["rule_id"] for r in active_rules].index(True)]["table_name"]
        if table_name not in table_metrics:
            table_metrics[table_name] = {
                "missing_rate": 0,
                "duplicate_rate": 0,
                "error_rate": 0,
                "inconsistent_rate": 0,
                "is_on_time": True # 此处对接数据产出监控系统获取实际值
            }
        # 按规则类型更新指标
        rule_type = active_rules[[r["rule_id"] == result["rule_id"] for r in active_rules].index(True)]["rule_type"]
        if rule_type == "完整性":
            table_metrics[table_name]["missing_rate"] = max(table_metrics[table_name]["missing_rate"], result["abnormal_rate"])
        elif rule_type == "唯一性":
            table_metrics[table_name]["duplicate_rate"] = max(table_metrics[table_name]["duplicate_rate"], result["abnormal_rate"])
        elif rule_type == "准确性":
            table_metrics[table_name]["error_rate"] = max(table_metrics[table_name]["error_rate"], result["abnormal_rate"])
        elif rule_type == "一致性":
            table_metrics[table_name]["inconsistent_rate"] = max(table_metrics[table_name]["inconsistent_rate"], result["abnormal_rate"])
    
    # 计算每个表的DQI
    dqi_results = []
    for table_name, metrics in table_metrics.items():
        dqi_score, level = calculate_dqi(metrics)
        dqi_results.append({
            "table_name": table_name,
            "dt": dt,
            "dqi_score": dqi_score,
            "quality_level": level,
            "metrics": metrics,
            "rule_results": [r for r in rule_results if active_rules[[rr["rule_id"] == r["rule_id"] for rr in active_rules].index(True)]["table_name"] == table_name]
        })
    
    # 保存结果到质量库
    dqi_df = spark.createDataFrame(dqi_results)
    dqi_df.write.mode("append").saveAsTable("dwd.dwd_data_quality_dqi_di")
    
    # 触发告警
    for result in dqi_results:
        if result["quality_level"] in ["合格", "不合格"]:
            send_alarm(result)
    
    return dqi_results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

apiVersion: apps/v1
kind: Deployment
metadata:
  name: data-quality-engine
spec:
  replicas: 3
  selector:
    matchLabels:
      app: data-quality-engine
  template:
    metadata:
      labels:
        app: data-quality-engine
    spec:
      containers:
      - name: engine
        image: data-quality-engine:v1.0.0
        ports:
        - containerPort: 8080
        resources:
          requests:
            cpu: "2"
            memory: "4Gi"
          limits:
            cpu: "4"
            memory: "8Gi"