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スキーマ・NULL・一意性のアサート

活用シーン

  1. 01データパイプラインの業務ルールを検証
  2. 02ETL 処理中にレコードを検査
  3. 03pytest テストにデータ品質アサートを追加

インストール

コマンドをコピーしてターミナルで実行します。

$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 対応