TensorFlow Data Validation如何获取包含异常类型的异常数据行？

结论 #

TensorFlow Data Validation（TFDV）原生不支持直接输出包含异常的具体数据行和对应异常类型，原因是tfdv.validate_statistics是基于数据集的统计特征进行校验，而非逐行扫描原始数据，仅会生成异常的汇总统计结果。

基于现有TFDV能力的自定义实现思路 #

你可以结合TFDV输出的异常规则，自行逐行校验原始DataFrame，生成你需要的结果格式，步骤如下：

1. 从TFDV生成的anomalies对象中提取所有预设的校验规则，比如示例中c1字段的取值范围为1~3
2. 遍历原始DataFrame的每一行，匹配所有异常规则，记录不符合规则的行和对应的异常类型
3. 把记录的结果整理为目标表格格式

示例实现代码 #

import pandas as pd
import tensorflow_data_validation as tfdv
from tensorflow_metadata.proto import schema_pb2

# 你的原始业务代码
df = pd.DataFrame({
    "c1": [1,2,3,100,100000],
    "c2": ["X","Y","X","Z","A"]
})
df_stats = tfdv.generate_statistics_from_dataframe(df)
schema = tfdv.infer_schema(statistics=df_stats)
tfdv.set_domain(schema, "c1", schema_pb2.IntDomain(min=1, max=3))
anomalies = tfdv.validate_statistics(statistics=df_stats, schema=schema)

# 自定义校验逻辑
# 第一步：整理所有字段的校验规则，可按需扩展字符串枚举、缺失值、类型校验等规则
rules = {}
for feature in schema.feature:
    if feature.type == schema_pb2.INT and feature.HasField("int_domain"):
        rules[feature.name] = {
            "type": "int_range",
            "min": feature.int_domain.min,
            "max": feature.int_domain.max,
            "err_msg": f"{feature.name} 取值超出范围"
        }

# 第二步：逐行校验
anomaly_rows = []
for idx, row in df.iterrows():
    row_anomalies = []
    for col, rule in rules.items():
        if rule["type"] == "int_range":
            if not (rule["min"] <= row[col] <= rule["max"]):
                row_anomalies.append(rule["err_msg"])
    if row_anomalies:
        anomaly_row = row.to_dict()
        anomaly_row["索引"] = idx
        anomaly_row["异常类型"] = "、".join(row_anomalies)
        anomaly_rows.append(anomaly_row)

# 第三步：生成目标格式结果
result_df = pd.DataFrame(anomaly_rows)[["索引", "c1", "c2", "异常类型"]]
print(result_df)

替代方案推荐 #

如果不想自行开发校验逻辑，可以使用专门做逐行数据校验的工具：

• Great Expectations：支持自定义丰富的校验规则，可直接输出所有异常行、异常原因，内置大量常用校验算子，也支持导出结构化结果表
• Pandera：轻量级的pandas数据校验库，可以直接和DataFrame结合使用，校验失败后可直接获取所有不符合规则的行及异常信息

内容的提问来源于stack exchange，提问作者datapug