data-quality-frameworks

データ品質フレームワーク

Great Expectations、dbt テスト、データコントラクトを用いてデータ品質を実装し、信頼性の高いデータパイプラインを確保するための本番パターンです。

このスキルを使用する場面

• パイプラインにデータ品質チェックを実装する場合
• Great Expectations の検証を設定する場合
• 包括的な dbt テストスイートを構築する場合
• チーム間でデータコントラクトを確立する場合
• データ品質メトリクスを監視する場合
• CI/CD でデータ検証を自動化する場合

中核となる概念

1. データ品質の次元

次元	説明	チェックの例
完全性	欠損値がない	expect_column_values_to_not_be_null
一意性	重複がない	expect_column_values_to_be_unique
妥当性	値が想定範囲内にある	expect_column_values_to_be_in_set
正確性	データが現実と一致する	クロスリファレンス検証
一貫性	矛盾がない	expect_column_pair_values_A_to_be_greater_than_B
適時性	データが最新である	expect_column_max_to_be_between

2. データのためのテストピラミッド

          /\
         /  \     Integration Tests (cross-table)
        /────\
       /      \   Unit Tests (single column)
      /────────\
     /          \ Schema Tests (structure)
    /────────────\

クイックスタート

Great Expectations のセットアップ

# Install
pip install great_expectations

# Initialize project
great_expectations init

# Create datasource
great_expectations datasource new

# great_expectations/checkpoints/daily_validation.yml
import great_expectations as gx

# Create context
context = gx.get_context()

# Create expectation suite
suite = context.add_expectation_suite("orders_suite")

# Add expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)

# Validate
results = context.run_checkpoint(checkpoint_name="daily_orders")

詳細なパターンと実例

詳細なパターンのドキュメントは references/details.md にあります。上記のナビゲーション階層では不十分な場合は、そのファイルを参照してください。

概要: {total_passed}/{total_tables} テーブルが合格")

    report.append("")

    for table, result in results.items():
        status = "✅" if result.passed else "❌"
        report.append(f"### {status} {table}")
        report.append(f"- Expectations: {result.total_expectations}")
        report.append(f"- Failed: {result.failed_expectations}")

        if not result.passed:
            report.append("- Failed checks:")
            for detail in result.details:
                if not detail["success"]:
                    report.append(f"  - {detail['expectation']}: {detail['observed_value']}")
        report.append("")

    return "\n".join(report)

Usage

context = gx.get_context() pipeline = DataQualityPipeline(context)

tables_to_validate = { "orders": "orders_suite", "customers": "customers_suite", "products": "products_suite", }

results = pipeline.run_all(tables_to_validate) report = pipeline.generate_report(results)

Fail pipeline if any table failed

if not all(r.passed for r in results.values()): print(report) raise ValueError("Data quality checks failed!")

## Best Practices

### Do's

- **Test early** - Validate source data before transformations
- **Test incrementally** - Add tests as you find issues
- **Document expectations** - Clear descriptions for each test
- **Alert on failures** - Integrate with monitoring
- **Version contracts** - Track schema changes

### Don'ts

- **Don't test everything** - Focus on critical columns
- **Don't ignore warnings** - They often precede failures
- **Don't skip freshness** - Stale data is bad data
- **Don't hardcode thresholds** - Use dynamic baselines
- **Don't test in isolation** - Test relationships too