工作原理
每个维度都需要具体测试。完整性可以是有非空客户 ID 的订单占比；准确性把数值与可信来源比较，例如支付结算文件；一致性检查同一客户在两个系统中是否属于同一分层；及时性衡量最新记录相对预期新鲜度的年龄；有效性检查格式和域规则，例如 ISO 国家代码；唯一性检查重复键。
只有当度量附带阈值和负责人时才有用。每次加载后运行的检查可以阻止发布或开启事件；趋势线能发现单个阈值漏掉的逐渐劣化。按来源或地区切分指标，因为全局平均值可能掩盖严重损坏的分区。
选择维度
从数据支持的决策出发。欺诈模型更需要及时性和准确性，而不是可选字段的完整性；财务报告需要准确性和一致性，可以容忍批处理延迟；客户名单需要唯一性和有效性。为每个用例记录“好”的含义，而不是最大化所有指标。
权衡
更多检查提高信心，但增加计算、告警噪音和管道延迟。精确的准确性检查通常需要第二个来源，而第二个来源可能不存在。阈值太紧会误报，太松则永远抓不到真问题。质量工作与功能交付竞争资源，因此应优先处理驱动收入、合规或客户信任的数据集。
示例与误解
一张订单表完整性为 99.9%，团队因此报告高质量。但有 12% 的行晚到超过六小时，破坏了当日履约。完整性没问题，失败的是及时性。增加一个带负责人的新鲜度检查，暴露了真正的问题。
误解是质量可以用一个分数概括。适用性取决于用例，不同维度之间还会相互权衡。
代码示例
-- 每个邮箱保留 created_at 最新的一行，标记其余为重复
WITH ranked AS (
SELECT
id,
email,
created_at,
ROW_NUMBER() OVER (
PARTITION BY email
ORDER BY created_at DESC, id DESC
) AS rn
FROM subscribers
)
SELECT id, email, created_at, (rn = 1) AS is_primary
FROM ranked
ORDER BY email, rn;
-- 一次性清扫：删除全部非首行
WITH ranked AS (
SELECT id,
ROW_NUMBER() OVER (
PARTITION BY email
ORDER BY created_at DESC, id DESC
) AS rn
FROM subscribers
)
DELETE FROM subscribers s
USING ranked r
WHERE s.id = r.id
AND r.rn > 1;
-- 上游堵漏：唯一索引阻止重复再次写入
CREATE UNIQUE INDEX CONCURRENTLY idx_subscribers_email
ON subscribers (email);package main
import (
"fmt"
"math/bits"
)
type Bitset []uint64
func NewBitset(size int) Bitset {
if size < 0 {
panic("negative bitset size")
}
return make(Bitset, (size+63)/64)
}
func (b Bitset) Set(index int) {
b[index/64] |= uint64(1) << (uint(index) % 64)
}
func (b Bitset) Clear(index int) {
b[index/64] &^= uint64(1) << (uint(index) % 64)
}
func (b Bitset) Has(index int) bool {
return b[index/64]&(uint64(1)<<(uint(index)%64)) != 0
}
func (b Bitset) Union(other Bitset) Bitset {
size := len(b)
if len(other) > size {
size = len(other)
}
out := make(Bitset, size)
for i := range out {
var left, right uint64
if i < len(b) {
left = b[i]
}
if i < len(other) {
right = other[i]
}
out[i] = left | right
}
return out
}
func (b Bitset) Intersect(other Bitset) Bitset {
size := len(b)
if len(other) < size {
size = len(other)
}
out := make(Bitset, size)
for i := range out {
out[i] = b[i] & other[i]
}
return out
}
func (b Bitset) Count() int {
total := 0
for _, word := range b {
total += bits.OnesCount64(word)
}
return total
}
func (b Bitset) Slice() []int {
out := make([]int, 0, b.Count())
for wordIndex, word := range b {
for word != 0 {
bit := bits.TrailingZeros64(word)
out = append(out, wordIndex*64+bit)
word &^= uint64(1) << uint(bit)
}
}
return out
}
func main() {
active := NewBitset(80)
for _, id := range []int{1, 5, 64, 79} {
active.Set(id)
}
blocked := NewBitset(80)
blocked.Set(5)
blocked.Set(9)
fmt.Println("count:", active.Count())
fmt.Println("has 64:", active.Has(64), "has 65:", active.Has(65))
fmt.Println("members:", active.Slice())
fmt.Println("union count:", active.Union(blocked).Count())
fmt.Println("intersect:", active.Intersect(blocked).Slice())
active.Clear(64)
fmt.Println("after clear:", active.Count())
}