本文目录导读：
在Python数据API中度量数据质量,通常从准确性、完整性、一致性、及时性和唯一性五个维度入手，下面我会结合Python生态（如Pandas、Great Expectations、Pydantic）给出具体的度量方法和代码示例。
基础数据质量维度与度量方法
1 完整性（Completeness）
度量：缺失值比例
import pandas as pd
def completeness_check(df, columns=None):
"""计算每列/指定列的完整性"""
if columns is None:
columns = df.columns
total_rows = len(df)
completeness = {}
for col in columns:
non_null_count = df[col].notna().sum()
completeness[col] = non_null_count / total_rows * 100
return pd.Series(completeness)
# 示例
data = pd.DataFrame({
'user_id': [1, 2, None, 4, 5],
'email': ['a@b.com', None, 'c@d.com', 'e@f.com', None]
})
print(completeness_check(data))
2 准确性（Accuracy）
度量：值与真实/预期值的匹配程度
import re
from datetime import datetime
def accuracy_check(df, rules):
"""
rules: dict of {column: validation_function}
返回每条记录的准确性得分
"""
results = {}
for col, func in rules.items():
results[col] = df[col].apply(func).mean() * 100
return results
# 示例规则
validation_rules = {
'email': lambda x: bool(re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', str(x))) if pd.notna(x) else False,
'age': lambda x: 0 <= x <= 150 if pd.notna(x) else False,
'date': lambda x: isinstance(x, datetime) if pd.notna(x) else False
}
print(accuracy_check(data, validation_rules))
3 一致性（Consistency）
度量：数据格式、单位或逻辑关系的统一性
def consistency_check(df, consistency_rules):
"""
检查字段间/字段内的一致性
如：开始日期 <= 结束日期
"""
issues = []
for rule_name, condition in consistency_rules.items():
violations = (~df.eval(condition)).sum()
issues.append({
'rule': rule_name,
'violations': violations,
'consistency_score': (len(df) - violations) / len(df) * 100
})
return pd.DataFrame(issues)
# 示例
df = pd.DataFrame({
'start_date': pd.to_datetime(['2023-01-01', '2023-02-01', '2023-03-01']),
'end_date': pd.to_datetime(['2023-01-10', '2023-01-15', '2023-02-20'])
})
rules = {
'date_range_valid': 'start_date <= end_date'
}
print(consistency_check(df, rules))
4 及时性（Timeliness）
度量：数据延迟、更新频率、新鲜度
from datetime import datetime, timedelta
def timeliness_check(df, date_column, max_lag_hours=24):
"""检查数据延迟是否在可接受范围内"""
now = datetime.now()
df['data_lag_hours'] = (now - df[date_column]).dt.total_seconds() / 3600
timely = (df['data_lag_hours'] <= max_lag_hours).mean() * 100
return {
'timeliness_score': timely,
'max_lag_hours': df['data_lag_hours'].max(),
'avg_lag_hours': df['data_lag_hours'].mean()
}
# 示例
df = pd.DataFrame({
'event_time': [datetime.now() - timedelta(hours=i*2) for i in range(10)]
})
print(timeliness_check(df, 'event_time', max_lag_hours=12))
5 唯一性（Uniqueness）
度量：重复记录比例
def uniqueness_check(df, key_columns):
"""检查关键字段的唯一性"""
total_records = len(df)
duplicates = df.duplicated(subset=key_columns, keep=False).sum()
return {
'total_records': total_records,
'duplicate_records': duplicates,
'uniqueness_score': (total_records - duplicates) / total_records * 100
}
# 示例
df = pd.DataFrame({
'user_id': [1, 2, 2, 4, 5], # ID为2的记录重复
'name': ['Alice', 'Bob', 'Bob', 'David', 'Eve']
})
print(uniqueness_check(df, ['user_id']))
整合的数据质量评分框架
class DataQualityMetrics:
def __init__(self, df):
self.df = df
self.results = {}
def calculate_completeness(self, columns=None):
cols = columns or self.df.columns
scores = {}
for col in cols:
non_null_pct = self.df[col].notna().mean() * 100
scores[col] = round(non_null_pct, 2)
self.results['completeness'] = scores
return scores
def calculate_uniqueness(self, key_columns):
dup_count = self.df.duplicated(subset=key_columns).sum()
score = (1 - dup_count / len(self.df)) * 100
self.results['uniqueness'] = round(score, 2)
return score
def calculate_accuracy(self, rules):
scores = {}
for col, func in rules.items():
valid = self.df[col].apply(func).sum()
score = (valid / len(self.df)) * 100
scores[col] = round(score, 2)
self.results['accuracy'] = scores
return scores
def calculate_timeliness(self, date_col, max_lag_hours=24):
lag = (datetime.now() - self.df[date_col]).dt.total_seconds() / 3600
timely_pct = (lag <= max_lag_hours).mean() * 100
self.results['timeliness'] = round(timely_pct, 2)
return timely_pct
def calculate_consistency(self, rules):
scores = {}
for rule_name, condition in rules.items():
valid = self.df.eval(condition).sum()
score = (valid / len(self.df)) * 100
scores[rule_name] = round(score, 2)
self.results['consistency'] = scores
return scores
def overall_score(self, weights=None):
"""计算综合质量评分"""
if weights is None:
weights = {
'completeness': 0.3,
'accuracy': 0.25,
'consistency': 0.2,
'timeliness': 0.15,
'uniqueness': 0.1
}
# 获取各维度平均分
avg_scores = {}
for dim in ['completeness', 'accuracy', 'consistency', 'timeliness', 'uniqueness']:
if dim in self.results:
if isinstance(self.results[dim], dict):
avg_scores[dim] = sum(self.results[dim].values()) / len(self.results[dim])
else:
avg_scores[dim] = self.results[dim]
total = sum(avg_scores.get(dim, 0) * weights.get(dim, 0) for dim in weights)
return round(total, 2)
# 使用示例
df = pd.DataFrame({
'user_id': [1, 2, 2, 4, 5],
'name': ['Alice', 'Bob', 'Bob', 'David', None],
'email': ['a@b.com', 'b@c.com', None, 'd@e.com', 'e@f.com'],
'created_at': [datetime.now() - timedelta(hours=i*3) for i in range(5)]
})
metrics = DataQualityMetrics(df)
# 逐维度计算
completeness = metrics.calculate_completeness()
uniqueness = metrics.calculate_uniqueness(['user_id'])
accuracy = metrics.calculate_accuracy({
'email': lambda x: bool(re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', str(x))) if pd.notna(x) else False
})
timeliness = metrics.calculate_timeliness('created_at', max_lag_hours=24)
print(f"Completeness: {completeness}")
print(f"Uniqueness: {uniqueness}")
print(f"Accuracy: {accuracy}")
print(f"Timeliness: {timeliness}")
print(f"Overall Score: {metrics.overall_score()}")
常用专业工具
1 Great Expectations（推荐）
import great_expectations as ge
# 将DataFrame转为Great Expectations格式
ge_df = ge.from_pandas(df)
# 定义期望/规则
ge_df.expect_column_values_to_not_be_null('user_id')
ge_df.expect_column_values_to_match_regex('email', r'^[\w\.-]+@[\w\.-]+\.\w+$')
ge_df.expect_column_values_to_be_unique('user_id')
# 生成数据质量报告
results = ge_df.validate()
print(results)
2 Pydantic（API输入验证）
from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional
class UserAPIInput(BaseModel):
user_id: int = Field(..., ge=0)
name: str = Field(..., min_length=1, max_length=100)
email: EmailStr
age: Optional[int] = Field(None, ge=0, le=150)
@validator('name')
def name_must_be_alpha(cls, v):
if not v.isalpha():
raise ValueError('Name must contain only letters')
return v
# 自动验证API输入数据质量
try:
user = UserAPIInput(**raw_data)
print("Data quality check passed")
except Exception as e:
print(f"Data quality violation: {e}")
实际应用建议
-
分阶段度量：
- 输入阶段：使用Pydantic/Schema验证
- 处理阶段：使用Pandas内置检查
- 输出阶段：使用Great Expectations生成报告
-
阈值设定：
- 关键字段（如ID）：完整性≥99.9%
- 业务字段（如金额）：准确性≥99%
- 辅助字段（如备注）：完整性≥80%
-
监控与告警：
def alert_if_data_quality_drops(quality_score, threshold=95): if quality_score < threshold: send_alert(f"Data quality score dropped to {quality_score}%") -
持续度量：
- 每次API调用时计算
- 定期生成质量趋势报告
- 建立数据质量基线
这样你就拥有了一套完整的Python数据API质量度量框架,根据实际业务场景，可以灵活调整各维度的权重和阈值。
