mcpField
D
data-validation

data-validation

github.com/majesticlabs-dev数据与数据库5080 次安装7 周前更新
代码仓库

关于 data-validation

面向构建验证流水线的数据工程师，提供使用自定义业务规则进行数据验证的技能。它包含可复用检查和基于 pandas 的流水线模式，并说明何时可考虑专用框架。

主要功能

  1. 01检查重复项和引用完整性
  2. 02验证日期范围与允许值
  3. 03组合检查构建验证流水线
  4. 04运行配置驱动的验证
  5. 05断言架构、空值和唯一性

适用场景

  1. 01验证数据流水线中的业务规则
  2. 02在 ETL 处理中检查记录
  3. 03在 pytest 测试中添加数据质量断言

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/majesticlabs-dev/majestic-marketplace --skill data-validation

安装后在 Claude Code 中输入/data-validation确认已加载。

附带文件scripts/validators.py

详细介绍

数据验证

适用对象： 构建验证流水线的数据工程师。

目标： 为自定义业务规则提供验证模式。

特定框架的技能：

  • pydantic-validation - 使用 Pydantic 进行记录级验证
  • pandera-validation - DataFrame 模式验证
  • great-expectations - 流水线预期和监控

脚本

从 scripts/validators.py 执行验证函数：

from scripts.validators import (
    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
)

框架选择

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

使用示例

基本验证
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')
验证流水线
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']}")
配置驱动的流水线
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)
测试断言
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'])

依赖项

pandas

评论

支持 Markdown