AI 数据治理 Agent：从被动发现问题到主动修复的升级路径
AI 数据治理 Agent：从被动发现问题到主动修复的升级路径
一、数据治理的"打地鼠"困境
做数据治理的人都有一个共同的痛：问题永远是事后才知道。
上周二早上9点，运营跑来问我："大喜，为什么昨天的 GMV 环比跌了 40%？"我心里一紧，开始排查——ETL 任务延迟了 3 小时导致分区数据没更新。问题是，这个延迟在凌晨 3 点就发生了，但没人知道，直到上班后才被发现。
这就是典型的被动数据治理：监控报警 → 人工排查 → 找到原因 → 手动修复 → 补数据。每一步都是人在驱动，数据质量靠"打地鼠"维持。
我们团队去年开始尝试用 AI Agent 来自动化数据治理，从"出问题再修"变成了"主动巡检 + 自动修复"，今天就分享这套升级路径。
flowchart LR
subgraph 传统模式["传统被动模式"]
T1[数据异常发生] --> T2[业务方发现] --> T3[人工排查] --> T4[手动修复]
end
subgraph Agent模式["AI Agent 主动模式"]
A1[定时巡检任务] --> A2{异常检测}
A2 -->|发现异常| A3[AI诊断根因]
A3 --> A4{可自动修复?}
A4 -->|是| A5[自动执行修复SQL]
A4 -->|否| A6[生成工单 + 修复建议]
A5 --> A7[自动验证修复结果]
A6 --> A8[通知值班人员]
end
style T4 fill:#E74C3C,color:#fff
style A5 fill:#27AE60,color:#fff
style A7 fill:#27AE60,color:#fff
二、数据治理 Agent 的三层能力
我们的 Agent 按照能力成熟度分了三层，每一层解决不同的问题：
第一层：巡检与发现
传统监控只能检测"这条 SQL 有没有报错"，但数据质量问题远不止报错。空分区、数据量突降、字段枚举值缺失——这些问题 SQL 本身不报错，但业务影响很大。
Agent 的巡检模块做了这些事：
# quality_checker.py — AI 数据质量巡检 Agent
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
class DataQualityAgent:
"""数据质量巡检 Agent：自动扫描并诊断数据问题"""
def __init__(self, db_connection, rules_config: dict):
self.db = db_connection
self.rules = rules_config # 巡检规则配置
def run_daily_check(self, target_date: str) -> List[Dict]:
"""每日巡检入口：执行所有注册的检查规则"""
issues = []
# 检查1：分区完整性
# 每个核心表都必须有当天的分区，漏分区的风险最高
for table in self.rules.get("partition_tables", []):
if not self._check_partition_exists(table, target_date):
issues.append({
"type": "missing_partition",
"table": table,
"partition": target_date,
"severity": "critical", # 缺少分区属于严重问题
"auto_fixable": True, # 可以自动补跑ETL
})
# 检查2：数据量波动
# 环比下降超过阈值可能是ETL中途挂掉，上涨超过阈值可能是重复跑
for table_config in self.rules.get("volume_check_tables", []):
row_count = self._get_row_count(table_config["name"], target_date)
yesterday_count = self._get_row_count(
table_config["name"],
self._prev_date(target_date)
)
if yesterday_count > 0:
change_pct = (row_count - yesterday_count) / yesterday_count
threshold = table_config.get("volume_threshold", 0.3)
if abs(change_pct) > threshold:
issues.append({
"type": "volume_anomaly",
"table": table_config["name"],
"partition": target_date,
"current_count": row_count,
"yesterday_count": yesterday_count,
"change_pct": round(change_pct * 100, 1),
"severity": "warning" if abs(change_pct) < 0.5 else "critical",
"auto_fixable": False, # 数据量异常需要人工判断原因
})
# 检查3：空值率检查
for col_config in self.rules.get("null_check_columns", []):
null_rate = self._get_null_rate(
col_config["table"],
col_config["column"],
target_date
)
if null_rate > col_config.get("threshold", 0.1):
issues.append({
"type": "high_null_rate",
"table": col_config["table"],
"column": col_config["column"],
"null_rate": round(null_rate * 100, 1),
"severity": "warning",
"auto_fixable": False,
})
return issues
第二层：根因诊断
发现问题是第一步，更重要的是找到根因。这是 AI 真正能发力的地方。
传统做法是数据工程师一个一个排查——看上游任务日志、检查数据源变更、对比历史数据。这个过程动辄半小时起。Agent 拿到问题列表后，会并行执行以下诊断：
- 上游依赖检查：查调度系统，看该表的父任务有没有异常
- 数据源变更检测：对比今天的 schema 和昨天是否一致（有没有删字段、改类型）
- 数据内容比对：抽样今天的样本分布和昨天的分布做相似度检测
- 历史相似问题匹配：从问题知识库里找是否有类似案例及解决方案
# root_cause.py — 根因诊断引擎
def diagnose_partition_missing(self, table: str, partition: str) -> str:
"""诊断分区缺失的根因"""
# 1. 查询上游 ETL 任务状态
upstream_tasks = self._get_upstream_tasks(table)
failed_tasks = [t for t in upstream_tasks if self._task_failed(t, partition)]
if failed_tasks:
# 上游任务失败是最常见的原因
return (
f"根因：上游ETL任务 {failed_tasks[0]['name']} "
f"在 {partition} 分区执行失败。\n"
f"失败日志摘要：{failed_tasks[0]['error_log'][:200]}\n"
f"建议：先修复 {failed_tasks[0]['name']}，再重跑 {table} 的分区。"
)
# 2. 检查数据源是否变更
source_table = self._get_source_table(table)
if source_table and not self._check_partition_exists(source_table, partition):
return (
f"根因：数据源表 {source_table} 缺少 {partition} 分区。\n"
f"可能原因：源头数据未产出，建议联系数据接入方确认。"
)
# 3. 未知原因，交给 AI 推理
return self._ai_diagnose(table, partition, "missing_partition")
第三层：自动修复
诊断出根因后，有些问题可以自动修复。我们目前支持这些场景：
| 问题类型 | 自动修复方式 | 成功率 |
|---|---|---|
| 分区缺失（上游任务失败已修复） | 自动触发重跑 | 95% |
| 空值率高（已知原因） | 自动填充默认值或打标签 | 80% |
| 数据重复 | 自动执行去重 SQL | 90% |
| 枚举值偏移（数据源格式变更） | 自动添加映射规则 | 60% |
自动修复的核心是修完必须验证。Agent 不会执行完之后拍拍屁股走人，它会再次运行巡检规则，确认问题已解决。如果没解决，自动升级为人工工单。
三、AI 的分析能力：从规则到推理
上面的规则检查只是基础。AI Agent 真正厉害的地方是能处理没有明确规则的问题：
场景：字段枚举值偏移。 你有一个"用户等级"字段，值的集合应该是 {VIP1, VIP2, VIP3, VIP4, VIP5}。突然有一天出现了 "VIP-1"、"VIP-2" 这种新格式。规则引擎很难捕捉——它不是空值，也没有报错。
AI Agent 的做法：
# enum_drift_detect.py — 枚举值漂移检测
def detect_enum_drift(self, table: str, column: str, partition: str) -> dict:
"""检测枚举字段的值域是否发生漂移（新值出现或旧值消失）"""
# 1. 获取今天和历史的值分布
today_values = set(self._get_distinct_values(table, column, partition))
baseline_values = self._get_baseline_values(
table, column,
lookback_days=7 # 用过去7天作为基线，既稳定又能反映最近变化
)
# 2. 计算差异
new_values = today_values - baseline_values
lost_values = baseline_values - today_values
if new_values or lost_values:
# 3. AI 分析新值的模式
analysis = self._ai_analyze_values(
table, column,
new_values=list(new_values),
lost_values=list(lost_values)
)
# AI 可能输出："VIP-1"和"VIP-2"很可能是"VIP1""VIP2"加了横线，
# 可能是上游系统的格式变更
return {
"table": table,
"column": column,
"new_values": list(new_values),
"lost_values": list(lost_values),
"ai_analysis": analysis,
"severity": "warning",
"suggestion": self._ai_generate_fix_suggestion(analysis)
}
return {"status": "normal"}
这种"值看起来像什么"的判断，正是大模型的强项。规则引擎需要你提前定义所有合法值，而 AI 可以用语义理解来判断"VIP-1"大概率就是"VIP1"的格式变体。
四、落地经验：Automation 的边界
做了快一年 Agent 治理，最重要的教训是：不是所有问题都适合自动修复。
我们的原则是：
- 可逆的、影响面小的 → 自动修复（如补跑一个分区）
- 不可逆的、影响面大的 → 自动生成方案，人工确认后再执行（如修改表结构）
- 不确定根因的 → 自动诊断 + 人工决策
另外还有两条血泪教训：
教训1：修复要有审计日志。 自动修复了什么东西、什么时候修的、修之前和修之后的数据是什么样的，全部记下来。否则出了问题时你连"Agent 干了什么"都搞不清楚。
教训2：灰度上线。 先在低优表上跑一个月，确认 Agent 不会"过度治疗"（比如把正常波动当异常修复），再推广到核心表。我们曾经 Agent 把周末的流量低谷判定为"数据异常"，自动补了一堆默认值——教训深刻。
五、总结
数据治理 Agent 的本质是把数据工程师的"排查-诊断-修复"流程自动化。它不等于完全无人值守，而是把人的精力从重复性排查中解放出来，去做更有价值的架构优化和治理设计。
升级路径也很清晰：
- 先做好巡检——让问题"被发现"而不是"被报告"
- 再做根因诊断——让 AI 帮你缩小排查范围
- 最后做自动修复——但一定要留人工确认的开关
一句话收尾：最好的数据治理，是你根本感觉不到它在治理。
我是朱大喜，一个被数据质量问题折磨到自建 Agent 的数据分析师。你的团队是怎么做数据治理的？还在靠"打地鼠"吗？评论区聊聊~
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
2
0- 0
已为社区贡献3条内容
所有评论(0)