数据质量评估：数据质量指标体系DAMA

可视化演示

双模式展示DAMA数据质量框架：六维雷达图呈现完整性准确性一致性及时性唯一性有效性的量化评估，加权评分模式展示六维权重分配与综合健康度计算及SLA/MDM/ETL治理组件。

六维质量体系——后厨质检清单

想象你是一家餐厅的主厨。每天清晨，供应商的货车停在厨房门口，你手里攥着一张“食材到货检查单”:够不够量、标签对不对、规格统不统一、新不新鲜、有没有重复送货、能不能直接下锅。六件事，任何一件偷懒，晚上就有一桌客人替你买单。数据世界的主厨是数据工程师与算法工程师，他们的食材是成千上万条记录，而他们手里的检查单，正是DAMA数据质量指标体系。

DAMA把“数据质量”这个抽象感觉拆成六个可计算的维度：完整性(Completeness)衡量字段有没有缺漏，准确性(Accuracy)衡量值本身错没错，一致性(Consistency)衡量不同数据源是否互相矛盾，及时性(Timeliness)衡量数据是否还在有效期内，唯一性(Uniqueness)衡量同一实体有没有被重复记录，有效性(Validity)衡量数据是否满足格式与业务规则。

六个维度各回答一类“数据毛病”:缺字段、错数值、相互打架、更新太慢、重复建档、不合规。单看任一维度都会误判整体健康度——某张表准确率98%,但重复率高达40%,模型照样学坏。因此DAMA将六维得分加权合成一个综合质量分，用于部门间、月份间、批次间的横向对比。权重不是拍脑袋定的，而应从业务目标反推：金融风控更看重准确与及时，精准营销更看重完整与有效。

检查单上的条目越细，后厨的差错就越少。DAMA六维的价值，不在于它多么“高级”,而在于把“凭感觉判断数据行不行”的模糊答案，变成六个具体、能打分、可追踪的体检项目——主厨终于不用靠尝一口来判断整车食材的好坏。

完整性度量——食材够不够

菜谱上写着“番茄、鸡蛋、葱、姜、蒜”五样，你打开冰箱发现少了两样。食材不够，再好的菜谱也只能干瞪眼。数据也一样：顾客表里三成人的手机号是空的，订单表里一半订单没有商品编码——字段缺失，意味着后续任何分析、建模、风控都无从下嘴。

完整性(Completeness)衡量的是“该有的信息有没有”。最直观的度量是非空率：统计某个关键字段非空的记录数占总记录数的比例。覆盖度则更细致地看“每一行记录是否至少包含一组必需字段”,比如一条订单记录至少要包含订单号、用户ID、商品ID和时间戳。

缺失检测要注意两类情况：一是“真缺失”,数据库里就是 NULL 或空字符串；二是“假填充”,系统用"0“、“未知”、”1900-01-01"占位。后者更隐蔽，会让完整性分数虚高，把问题掩盖到下游。工程上通常逐字段扫描空值、空串、默认占位符，并记录缺失模式。

就像主厨每天开市前都要盘一遍冷藏库，DAMA完整性检查也是例行公事——单缺一味料还能临时调换，缺了主料就得当机立断改菜单，才不会等到客人催菜才手忙脚乱。

准确性度量——标签是真的吗

盐罐子外面贴着“盐”,里头装的却是糖——这道菜怎么做都是甜的。数据准确性就是“标签与实物是否相符”:字段值写的是“男性”,实际是一位女顾客；商品价格标的是 9.9,结账时却变成 19.9。

准确性(Accuracy)衡量字段值与真实世界的一致程度。常用度量是错误率：抽样或全量校验后，得出通过校验的记录占比。校验手段有三类：值域校验（数值是否在合理区间）、格式校验（手机号是否 11 位、邮箱是否包含 @）、引用校验（外键是否能对上一个确实存在的主键记录）。

准确性通常是六维中最难高分的维度，因为“真实值”往往需要人工核对或权威源比对。实际项目中，可以把准确性拆成“静态准确性”（存量数据抽查）和“动态准确性”（新写入数据实时校验）,前者用抽样审计，后者用规则引擎在写入链路中拦截。

巧妇难为无米之炊，更难为“假米”之炊。准确性之所以放在第二关，是因为它直接决定模型能不能学到真规律——标签错一半，训练出来的模型就像被误导的帮厨，越卖力越离谱。

一致性度量——标准统一吗

同一个连锁品牌，总店规定红烧肉用老抽，分店厨师却擅自换了生抽，顾客一吃就知道不是那个味儿。数据世界里，同一个客户在 CRM 系统里注册地址是“北京市朝阳区”,在订单系统里却写成“北京朝阳”,两条记录描述的是同一个实体，内容却互不相同。

一致性(Consistency)衡量同一数据在不同系统、不同表、不同时点之间是否保持一致。业务上最常见的检测方法是跨源比对：以某个权威源（如主数据表）为基准，统计其他源中与基准矛盾的记录数。矛盾检测还包括同一记录内字段自洽（如出生日期与年龄是否匹配）。

数据不一致的根源，通常是各系统各自维护一份字典表、单位表或地址库，缺少统一的主数据管理。解决方向是规范统一：定义企业级数据字典、统一编码和单位换算规则，让各系统在写入时遵循同一套标准，而不是事后反复“拉架”。

后厨统一调料的牌子，比每道菜都好吃更重要；数据治理统一了“度量衡”,跨系统联动的 AI 应用才能少吵架、少返工。

及时性度量——食材新鲜吗

冰柜里的牛肉虽然没变质，但过了最佳食用期，煎出来又柴又老。餐厅不敢用过期食材，AI 也不敢用“过期数据”。用户上周就卸载了 App,推荐系统却还在给他推会员续费券——这就是数据不及时带来的尴尬。

及时性(Timeliness)衡量数据从业务发生到可被查询使用的时间差是否在可接受范围内。核心度量是时效窗口（如“订单数据需在 5 分钟内入库”）和新鲜度（数据最近一次更新的时间距今多久）。工程上记录每条记录的入库时间戳，统计落在窗口内的记录比例，即延迟率越低、新鲜度越高。

不同业务对时效的容忍度差异很大：风控反欺诈要求秒级，经营日报允许小时级，离线训练集甚至可以接受 T+1。因此及时性阈值必须按数据用途分别设定，而不是一刀切用“越快越好”——追求过度实时会增加计算成本，却换不来业务收益。

主厨每天早上会检查所有食材的入库日期贴签，把临期食材优先排进菜单，超期的直接下架。数据团队也应当给每条数据打上“生产日期”,让陈旧数据在流入模型前就被拦下，而不是等模型给出离谱预测后才回头翻冰箱。

唯一性度量——重复买了吗

厨房里同一款土豆，今天上午采购了一箱，下午另一个采购又提回一箱，占了仓位不说，还容易过期浪费。数据表里一模一样的问题：同一个客户，因为线下填单和线上注册走了两个通道，被建了两条记录，后续营销短信就发了双份，模型也把他当成了两个人。

唯一性(Uniqueness)衡量同一实体在数据集中是否只有一条标准记录。核心度量是重复率：按照主键或业务键（身份证号、手机号、统一社会信用代码）做分组，统计重复组中的记录数占比。若无天然主键，则用一组“业务唯一键”（如姓名+手机号+注册时间）进行模糊匹配识别近似重复。

主键管理是唯一性的制度保障：在表结构层面为主键加唯一约束，在写入链路做插入前查重(upsert),并用数据清洗工具做周期去重。去重策略要谨慎——简单按时间戳保留最新一条，可能丢掉历史行为轨迹，更稳妥的做法是“合并档案”:把多条重复记录的证据合并成一条主记录，并在明细中保留历史版本。

后厨的采购台账，每类食材只有一行记录，领用、补货都挂在这行下面。数据只有“一本账”,AI 才不会把同一个人分裂成两个用户去画像。

有效性度量——能直接下锅吗

菜谱里写“糖 5 克”,你抓起一把糖愣是没称——“一把”不是菜谱认可的计量单位。数据的有效性也一样：字段里填的内容，必须是系统或业务规则认可的值。一个“正常”的值，未必是“有效”的值：年龄写成 999,手机号写成“abc“,状态字段写成”已完单“（实际字典里只有”已完成“和”已取消”）。

有效性(Validity)衡量数据是否符合预定义的规则，包括三类：格式规则（日期必须是 YYYY-MM-DD、邮箱必须含 @）、值域规则（性别 ∈ {男，女，未知}、折扣率 ∈ [0,1]）、业务规则（发货日期不早于下单日期、库存不得为负）。有效性在六维中最容易被自动化，只需把规则翻译成引擎里的校验函数。

规则引擎上线后要维护一份“规则清单”,每条规则标明归属和责任人，规则变更要走评审，否则旧数据按旧规则、新数据按新规则，历史对比就会失真。有效性得分低时，不要急着改数据，先判断是规则过严还是数据真错——这两者的修复方式完全不同。

主厨在食材入场时的最后一道关卡，是确认食材的规格符合菜谱要求——不能拿“海盐”冒充“普通盐”,不能拿“奶油”冒充“黄油”。数据也是一样，合规、合规格，才有资格下锅。

评估实施——天天做的质检

聪明的餐厅不会只在开业第一天检查食材，而是把质检变成每天的固定仪式：进货时查、营业前查、周会上对比这周和上周的食材损耗率。数据质量评估同样不是一次性项目，而是一套“定阈值—打分—监控—修复”的持续闭环。

阈值设定是实施的第一步：每个维度都定义一个“红线”。红线应来自业务容忍度，比如风控模型要求准确性≥98%、及时性≥95%,而分析报表可以放宽到准确性≥95%、及时性≥90%。评分模型把六维加权的综合分映射为质量等级，让管理层一眼看懂当前数据健康度，而不是面对六个小数不知所措。

监控与修复是闭环的后半段：质量分随数据管线每日计算，用趋势图追踪劣化苗头；某个维度跌破阈值时触发告警并定位到具体表、具体批次，责任工程师修复后重新评分。评分只有在跑批中沉淀成历史曲线，才真正成为“数据健康的体温计”。

一张六维雷达图挂在大屏上，就像后厨墙上的食材质量看板——哪一项亮了红灯，当天就处理哪一项。数据质量不是某个团队的负担，而是整条数据管线的共同习惯。养成这个习惯，AI 模型才能在干净的食材上做出真正的“米其林”。

AI 应用场景

Python 演示

"""DAMA六维数据质量评估器:对字典列表计算六大维度评分。"""from datetime import datetime, timedelta  class DataQualityChecker:    """输入一组记录,逐维计算质量得分。"""     def __init__(self, records, required_fields=None):        self.records = records        self.required_fields = required_fields or ["id", "name", "phone", "updated_at"]        self.completeness = 0.0        self.accuracy = 0.0        self.consistency = 0.0        self.timeliness = 0.0        self.uniqueness = 0.0        self.validity = 0.0     def evaluate_completeness(self):        total = len(self.records)        missing = 0        for row in self.records:            for field in self.required_fields:                if not row.get(field):                    missing += 1                    break        self.completeness = (total - missing) / total * 100 if total else 0.0        return self.completeness     def evaluate_accuracy(self):        total = len(self.records)        correct = 0        for row in self.records:            phone = row.get("phone", "")            if phone.isdigit() and len(phone) == 11:                correct += 1        self.accuracy = correct / total * 100 if total else 0.0        return self.accuracy     def evaluate_consistency(self):        total = len(self.records)        conflict = 0        seen = {}        for row in self.records:            uid = row.get("id")            city = row.get("city")            if uid in seen and seen[uid] != city:                conflict += 1            seen[uid] = city        self.consistency = (total - conflict) / total * 100 if total else 0.0        return self.consistency     def evaluate_timeliness(self):        total = len(self.records)        fresh = 0        deadline = datetime.now() - timedelta(days=7)        for row in self.records:            try:                updated = datetime.strptime(row.get("updated_at", ""), "%Y-%m-%d")                if updated >= deadline:                    fresh += 1            except ValueError:                pass        self.timeliness = fresh / total * 100 if total else 0.0        return self.timeliness     def evaluate_uniqueness(self):        total = len(self.records)        ids = [row.get("id") for row in self.records]        dup = total - len(set(ids))        self.uniqueness = (total - dup) / total * 100 if total else 0.0        return self.uniqueness     def evaluate_validity(self):        total = len(self.records)        valid_emails = 0        for row in self.records:            email = row.get("email", "")            if "@" in email and email.endswith(".com"):                valid_emails += 1        self.validity = valid_emails / total * 100 if total else 0.0        return self.validity     def overall(self):        return sum([            self.completeness, self.accuracy, self.consistency,            self.timeliness, self.uniqueness, self.validity        ]) / 6  def main():    sample = [        {"id": 1, "name": "陈晨", "phone": "13800138000", "city": "上海",         "email": "chen@demo.com", "updated_at": "2025-05-01"},        {"id": 2, "name": "李雷", "phone": "13900139000", "city": "北京",         "email": "li@demo.com", "updated_at": "2025-05-02"},        {"id": 1, "name": "陈晨", "phone": "13800138000", "city": "杭州",         "email": "chen@demo.com", "updated_at": "2023-01-10"},        {"id": 3, "name": "", "phone": "abcd", "city": "广州",         "email": "han@demo.com", "updated_at": "2025-04-25"},        {"id": 4, "name": "王芳", "phone": "13700137000", "city": "深圳",         "email": "bad-email", "updated_at": "2025-04-30"},    ]    checker = DataQualityChecker(sample)    print("六维质量评分:")    print("完整性: %.1f%%" % checker.evaluate_completeness())    print("准确性: %.1f%%" % checker.evaluate_accuracy())    print("一致性: %.1f%%" % checker.evaluate_consistency())    print("及时性: %.1f%%" % checker.evaluate_timeliness())    print("唯一性: %.1f%%" % checker.evaluate_uniqueness())    print("有效性: %.1f%%" % checker.evaluate_validity())    print("综合: %.1f%%" % checker.overall())  if __name__ == "__main__":    main()

1. from datetime import datetime, timedelta导入时间工具，用于判断数据更新是否在 7 天时效窗口内
2. if not row.get(field):字段为空值或空字符串时，把这条记录标记为缺失
3. self.completeness = (total - missing) / total * 100 if total else 0.0非空占比即完整性得分，同时防止除零
4. if phone.isdigit() and len(phone) == 11:手机号必须全是数字且恰好 11 位，否则判错
5. seen = {}` 配合 `if uid in seen and seen[uid] != city:同一个 id 出现在两个不同城市，就是一致性矛盾
6. deadline = datetime.now() - timedelta(days=7)把“时效窗口”翻译成具体的日期阈值
7. dup = total - len(set(ids))用集合天然去重，总记录数减集合大小就是重复数量
8. if "@" in email and email.endswith(".com")邮箱必须含 @ 且以 .com 结尾，是典型的有效性规则

常见误区

面试题