data-validation

用途

Data Validation 面向构建校验流水线的数据工程师，目标是提供针对自定义业务规则的校验模式。它既包含可直接调用的校验脚本，也按框架细分出多个子技能，帮助在不同数据场景下选择合适的校验方式。

主要能力

该技能提供脚本 scripts/validators.py，可从中导入以下内容：

ValidationResult
DataValidator
validate_no_duplicates
validate_referential_integrity
validate_date_range
validate_value_in_set
run_validation_pipeline
validate_with_schema_version
assert_schema_match
assert_no_nulls
assert_unique
assert_values_in_set

同时提供按框架划分的子技能：

pydantic-validation：使用 Pydantic 进行记录级校验
pandera-validation：DataFrame 模式校验
great-expectations：流水线期望与监控

使用方式

基础校验

可从 scripts.validators 导入 validate_no_duplicates 与 validate_referential_integrity。例如检查重复值：

from scripts.validators import validate_no_duplicates, validate_referential_integrity

# Check duplicates
result = validate_no_duplicates(df, cols=['id'])
if not result.passed:
    print(f"Error: {result.message}")
    print(result.failed_rows)

# Check referential integrity
result = validate_referential_integrity(df, 'user_id', users_df, 'id')

校验流水线

使用 DataValidator 添加多个校验检查：

from scripts.validators import DataValidator, validate_no_duplicates, validate_date_range

validator = DataValidator()
validator.add_check(lambda df: validate_no_duplicates(df, ['id']))
validator.add_check(lambda df: validate_date_range(df, 'created_at', '2020-01-01', '2025-12-31'))

results = validator.validate(df)
if not results['passed']:
    for check in results['checks']:
        if not check['passed']:
            print(f"Failed: {check['message']}")

配置驱动的流水线

使用 run_validation_pipeline 通过配置字典运行校验：

from scripts.validators import run_validation_pipeline

config = {
    'unique_columns': ['id'],
    'date_ranges': {
        'created_at': ('2020-01-01', '2025-12-31'),
        'updated_at': ('2020-01-01', '2025-12-31')
    }
}

clean_df, results = run_validation_pipeline(df, config)

测试断言

可在 pytest 中使用断言函数：

from scripts.validators import assert_schema_match, assert_no_nulls, assert_unique

# In pytest
def test_data_quality():
    assert_schema_match(df, {'id': 'int64', 'email': 'object'})
    assert_no_nulls(df, ['id', 'email'])
    assert_unique(df, ['id'])

框架选择

使用场景	框架
API 请求/响应	Pydantic
逐条 ETL	Pydantic
DataFrame 校验	Pandera
DataFrame 类型提示	Pandera
流水线监控	Great Expectations
数据仓库检查	Great Expectations
自定义业务规则	自定义函数（本技能）

适用场景

适用于需要为数据管道添加自定义业务规则校验的数据工程场景，包括重复值检查、引用完整性检查、日期范围校验、值集合校验、模式版本校验，以及在 pytest 中编写数据质量测试断言。

依赖

pandas

限制

资料未提供关于收费、价格、平台支持范围或具体效果的信息。