ストーリー

データ品質の6つの次元

基本の6次元

各次元の測定方法

from dataclasses import dataclass
from typing import Optional

@dataclass
class DataQualityMetrics:
    """データ品質の6次元を測定する"""

    # 正確性: 既知の正解データとの一致率
    accuracy_score: float  # 0.0 - 1.0

    # 完全性: NULL/欠損でないレコードの割合
    completeness_score: float  # 0.0 - 1.0

    # 一貫性: システム間で矛盾のないレコードの割合
    consistency_score: float  # 0.0 - 1.0

    # 適時性: 鮮度基準を満たすレコードの割合
    timeliness_score: float  # 0.0 - 1.0

    # 有効性: バリデーションルールに適合するレコードの割合
    validity_score: float  # 0.0 - 1.0

    # 一意性: 重複のないレコードの割合
    uniqueness_score: float  # 0.0 - 1.0

    @property
    def overall_score(self) -> float:
        """加重平均による総合品質スコア"""
        weights = {
            "accuracy": 0.25,
            "completeness": 0.20,
            "consistency": 0.20,
            "timeliness": 0.15,
            "validity": 0.10,
            "uniqueness": 0.10,
        }
        return (
            self.accuracy_score * weights["accuracy"]
            + self.completeness_score * weights["completeness"]
            + self.consistency_score * weights["consistency"]
            + self.timeliness_score * weights["timeliness"]
            + self.validity_score * weights["validity"]
            + self.uniqueness_score * weights["uniqueness"]
        )

測定例

顧客データ品質レポート
┌────────────┬───────┬─────────────────────┐
│ 次元       │ スコア │ 基準                │
├────────────┼───────┼─────────────────────┤
│ 正確性     │ 0.92  │ 住所照合サービスで検証 │
│ 完全性     │ 0.87  │ 必須フィールドの充足率 │
│ 一貫性     │ 0.78  │ CRM-DWH間の不一致率   │
│ 適時性     │ 0.95  │ 24h以内更新の割合     │
│ 有効性     │ 0.99  │ スキーマ適合率        │
│ 一意性     │ 0.96  │ 重複検出率           │
├────────────┼───────┼─────────────────────┤
│ 総合スコア │ 0.91  │ AI利用基準: 0.80以上  │
└────────────┴───────┴─────────────────────┘

AI向けデータ品質の特殊要件

従来のデータ品質に加え、AIシステムには固有の品質要件があります。

1. バイアス検出

from typing import Dict, List

def detect_demographic_bias(
    dataset: list[dict],
    target_column: str,
    sensitive_columns: list[str],
) -> dict[str, dict]:
    """
    データセットの人口統計バイアスを検出する

    Args:
        dataset: 分析対象データセット
        target_column: 予測対象のカラム
        sensitive_columns: 保護属性のカラム（性別、年齢層等）

    Returns:
        各保護属性ごとのバイアスメトリクス
    """
    bias_report: dict[str, dict] = {}

    for col in sensitive_columns:
        # 各グループの分布を計算
        group_distributions = {}
        for record in dataset:
            group = record[col]
            target = record[target_column]
            if group not in group_distributions:
                group_distributions[group] = {"positive": 0, "total": 0}
            group_distributions[group]["total"] += 1
            if target == 1:  # positive class
                group_distributions[group]["positive"] += 1

        # Disparate Impact Ratio の計算
        rates = {
            g: d["positive"] / d["total"]
            for g, d in group_distributions.items()
            if d["total"] > 0
        }
        if rates:
            min_rate = min(rates.values())
            max_rate = max(rates.values())
            disparate_impact = min_rate / max_rate if max_rate > 0 else 0

            bias_report[col] = {
                "group_rates": rates,
                "disparate_impact_ratio": disparate_impact,
                "is_biased": disparate_impact < 0.8,  # 4/5ルール
            }

    return bias_report

2. データドリフト検出

from scipy import stats
import numpy as np

def detect_data_drift(
    reference_data: np.ndarray,
    current_data: np.ndarray,
    threshold: float = 0.05,
) -> dict:
    """
    KS検定によるデータドリフトの検出

    Args:
        reference_data: ベースライン（学習時）のデータ分布
        current_data: 現在のデータ分布
        threshold: p値の閾値

    Returns:
        ドリフト検出結果
    """
    ks_statistic, p_value = stats.ks_2samp(reference_data, current_data)

    return {
        "ks_statistic": ks_statistic,
        "p_value": p_value,
        "drift_detected": p_value < threshold,
        "severity": (
            "HIGH" if ks_statistic > 0.2
            else "MEDIUM" if ks_statistic > 0.1
            else "LOW"
        ),
        "recommendation": (
            "モデルの再学習を推奨"
            if p_value < threshold
            else "現行モデルで継続可能"
        ),
    }

3. ラベル品質

データプロファイリングと品質スコアリング

データプロファイリングのプロセス

# データプロファイリング実行定義
data_profiling:
  schedule: "daily at 02:00 JST"
  targets:
    - dataset: "customer_profile"
      checks:
        - type: "completeness"
          columns: ["name", "email", "phone"]
          threshold: 0.95
        - type: "uniqueness"
          columns: ["customer_id", "email"]
          threshold: 0.999
        - type: "validity"
          rules:
            email: "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"
            phone: "^\\d{2,4}-\\d{2,4}-\\d{4}$"
          threshold: 0.98
        - type: "freshness"
          column: "updated_at"
          max_age: "24h"
        - type: "statistical"
          columns: ["age", "purchase_amount"]
          checks: ["mean", "stddev", "min", "max", "percentiles"]

    - dataset: "product_catalog"
      checks:
        - type: "completeness"
          columns: ["product_id", "name", "price", "description"]
          threshold: 0.99
        - type: "consistency"
          cross_reference:
            source: "product_catalog.category_id"
            target: "category_master.id"
          threshold: 1.0

品質スコアリングの自動化

Great Expectations を用いた品質チェックの実装例です。

import great_expectations as gx

# データコンテキストの作成
context = gx.get_context()

# Expectation Suite の定義
suite = context.add_expectation_suite("customer_data_quality")

# 品質ルールの定義
expectations = [
    # 完全性
    {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {"column": "customer_id"},
    },
    {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {"column": "email", "mostly": 0.95},
    },
    # 有効性
    {
        "type": "expect_column_values_to_match_regex",
        "kwargs": {
            "column": "email",
            "regex": r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
            "mostly": 0.98,
        },
    },
    # 一意性
    {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {"column": "customer_id"},
    },
    # 範囲チェック
    {
        "type": "expect_column_values_to_be_between",
        "kwargs": {"column": "age", "min_value": 0, "max_value": 150},
    },
    # 参照整合性
    {
        "type": "expect_column_values_to_be_in_set",
        "kwargs": {
            "column": "status",
            "value_set": ["active", "inactive", "suspended"],
        },
    },
]

データクレンジング戦略

クレンジングのレベル

AI向けクレンジングのベストプラクティス

class AIDataCleansingPipeline:
    """AI向けデータクレンジングパイプライン"""

    def execute(self, dataset):
        # Step 1: 重複除去
        dataset = self.deduplicate(dataset, strategy="fuzzy_matching")

        # Step 2: 欠損値処理
        dataset = self.handle_missing(
            dataset,
            strategy="domain_aware",  # ドメイン知識に基づく補完
            # AI学習用は「欠損」を明示するのが安全
            # 安易な平均値補完はバイアスを生む
        )

        # Step 3: 異常値処理
        dataset = self.handle_outliers(
            dataset,
            method="isolation_forest",
            action="flag",  # 削除ではなくフラグを付ける
        )

        # Step 4: 正規化
        dataset = self.normalize(
            dataset,
            text_columns=["description", "comment"],
            # Unicode正規化、全角半角統一
        )

        # Step 5: PII検出とマスキング
        dataset = self.detect_and_mask_pii(
            dataset,
            pii_types=["email", "phone", "name", "address"],
        )

        # Step 6: バイアスチェック
        bias_report = self.check_bias(
            dataset,
            sensitive_columns=["gender", "age_group", "region"],
        )

        return dataset, bias_report

データ品質SLOの設定方法

SLOの設計原則

データ品質SLOの設定例

data_quality_slos:
  customer_profile:
    target: "AI推薦エンジンの精度維持"
    slos:
      - metric: "overall_quality_score"
        target: 0.85
        window: "7d rolling"
        error_budget: "7日間で品質スコアが基準を下回る日数 <= 1日"

      - metric: "completeness"
        target: 0.95
        columns: ["name", "email", "purchase_history"]
        window: "daily"

      - metric: "freshness"
        target: "< 24h"
        column: "updated_at"
        window: "daily"

      - metric: "bias_score"
        target: "disparate_impact_ratio >= 0.8"
        sensitive_columns: ["gender", "age_group"]
        window: "monthly"

    alerts:
      - condition: "overall_quality_score < 0.80"
        severity: "critical"
        action: "RAGパイプラインを一時停止し、データスチュワードに通知"

      - condition: "completeness < 0.90"
        severity: "warning"
        action: "データエンジニアに改善チケット起票"

      - condition: "bias_score.disparate_impact_ratio < 0.8"
        severity: "critical"
        action: "AI利用を停止し、データオーナーに即座にエスカレーション"

品質ダッシュボードの構成

┌──────────────────────────────────────────────┐
│ データ品質ダッシュボード                        │
├──────────────────────────────────────────────┤
│                                              │
│  総合品質スコア: 87/100  [===========---]     │
│                                              │
│  ┌─────────────────┬─────────────────┐       │
│  │ 正確性: 92%  ✓  │ 完全性: 87%  ▲  │       │
│  ├─────────────────┼─────────────────┤       │
│  │ 一貫性: 78%  ▼  │ 適時性: 95%  ✓  │       │
│  ├─────────────────┼─────────────────┤       │
│  │ 有効性: 99%  ✓  │ 一意性: 96%  ✓  │       │
│  └─────────────────┴─────────────────┘       │
│                                              │
│  AI固有指標:                                  │
│  ├ バイアススコア: 0.85 (DI ratio)  ✓        │
│  ├ ドリフト: 未検出  ✓                       │
│  └ ラベル品質 (Kappa): 0.82  ✓               │
│                                              │
│  SLO状態: 7日間エラーバジェット残: 86%         │
│                                              │
└──────────────────────────────────────────────┘

まとめ

チェックリスト

• データ品質の6つの次元を列挙し、それぞれ説明できる
• AI向けの特殊な品質要件（バイアス、ドリフト、ラベル品質）を理解した
• データプロファイリングの仕組みと主要ツールを把握した
• クレンジング戦略の4レベルを区別できる
• データ品質SLOの設計原則を理解した

次のステップへ

次は「プライバシーとコンプライアンス」を学びます。AIシステムにおける個人情報保護と法規制対応の実践的な方法を深掘りしましょう。

推定読了時間: 30分