売上データが吹っ飛んだ夜から始めたデータ品質管理の実装記録

先日、うちのチームで悪夢みたいな出来事があった。

夜中11時、Slackに「ダッシュボードの売上が先月比150%超えてる、バグじゃないか？」という通知が届いた。調べてみたら、上流のETLパイプラインで通貨の単位変換処理がコケていて、円建てとドル建てが混在したデータがRedshiftに流れ込んでいた。その状態で集計が走り、BIに「売上爆増」として表示されていた。

実害は翌朝の役員会議で間違った数字をもとにした意思決定が行われる手前で止まったけど、発覚が6時間遅れていたら本当にまずかった。本番で売上データが150%跳ね上がった日、データ品質管理と向き合った話でも触れたように、データ品質の問題は「気づかないまま信頼してしまう」のが一番怖い。

その事件以来、チームで本格的にデータ品質管理の仕組みを整え直した。2026年現在の最新ツール事情も交えながら、実際に動かしてみた知見を書く。

なぜデータ品質問題は検知が遅れるのか

そもそもの話をすると、データの品質問題がアプリケーションのバグより発見しにくい理由がある。アプリが落ちれば即座にアラートが飛ぶ。でもデータの問題は「動いているように見える」。パイプラインは正常終了しているし、ダッシュボードも表示されている。ただ、中身がおかしい。

flowchart TD
    A[上流システム\nDB/API/ファイル] --> B[ETL/ELTパイプライン]
    B --> C{データ品質チェック}
    C -->|品質OK| D[Data Warehouse\nRedshift/BigQuery]
    C -->|品質NG| E[アラート通知\nSlack/PagerDuty]
    C -->|品質NG| F[隔離ストレージ\nQuarantine Layer]
    D --> G[BIツール\nMetabase/Superset]
    D --> H[ML/分析基盤]
    E --> I[エンジニア対応]
    F --> I
    style C fill:#ff9800,color:#fff
    style E fill:#f44336,color:#fff
    style F fill:#9c27b0,color:#fff

うちのチームで起きていた問題を整理すると、チェックポイントが存在していなかった。パイプラインは動くが品質検証がなく、異常データがそのままDWHに流れ込む構成になっていたわけだ。

2026年時点で、データ品質管理のツール選択肢はだいぶ成熟してきた。主要ツールを整理するとこんな感じ。

ツール	カテゴリ	料金モデル	2026年の特徴
dbt test / dbt-expectations	OSS・変換レイヤー	無料（Core）	v2.0でテスト構文が大幅強化
Great Expectations (GX Cloud)	OSS＋SaaS	GX Cloud有料	2025年末にAI自動期待値生成機能追加
Monte Carlo	SaaS	有料	LLMベース異常検知が成熟
Soda Core	OSS＋SaaS	Soda Cloud有料	YAML駆動で学習コスト低め
AWS Glue Data Quality	マネージド	使用量課金	Glue 5.0と統合
Bigeye	SaaS	有料	統計的ベースライン自動生成が強い

正直、全部試したわけじゃないけど、チームではdbt test + Soda Core + Monte Carloの組み合わせに落ち着いた。理由は後述する。

dbt testで「変換直後」の品質を担保する

最初に入れたのがdbt testだった。dbt Core 2.0以降、テスト構文がかなり整理されてdbt-expectations拡張も安定してきた。dbt Core 2.0移行後1年、本番運用で積み上がった設計の知見でも書いたけど、2.0からはUnit Test機能がモデルレベルで使えるようになっている。

実際にうちで使っているテスト定義を見てほしい。

# models/marts/finance/fct_sales.yml
models:
  - name: fct_sales
    description: "売上ファクトテーブル（円建て統一）"
    columns:
      - name: sale_id
        data_tests:
          - unique
          - not_null
      - name: amount_jpy
        data_tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 100000000  # 1億円を上限に（異常値フラグ）
          - dbt_expectations.expect_column_value_z_scores_to_be_less_than:
              threshold: 4.0  # 標準偏差4以上を外れ値として検知
      - name: currency_code
        data_tests:
          - accepted_values:
              values: ['JPY']  # 円建て以外が入ってきたら即エラー
      - name: sale_date
        data_tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: "'2020-01-01'"
              max_value: "current_date"

# ソースレベルでのフレッシュネスチェック
sources:
  - name: raw_transactions
    freshness:
      warn_after: {count: 6, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: _loaded_at

これだけで、通貨コードがJPY以外のレコードが入ってきた瞬間にdbt runが失敗する。あの事件が起きた時にこれがあれば、少なくともモデルビルドの段階で止まっていたはずだ。

Unit Testの書き方も見ておこう。

# models/marts/finance/fct_sales.yml (unit test追記)
unit_tests:
  - name: test_currency_normalization
    description: "USD→JPYの通貨変換が正しく動作すること"
    model: fct_sales
    given:
      - input: ref('stg_transactions')
        rows:
          - {transaction_id: 1, amount: 100, currency: 'USD', rate: 149.5}
          - {transaction_id: 2, amount: 5000, currency: 'JPY', rate: 1.0}
    expect:
      rows:
        - {sale_id: 1, amount_jpy: 14950, currency_code: 'JPY'}
        - {sale_id: 2, amount_jpy: 5000, currency_code: 'JPY'}

このUnit Test機能、最初は「dbtでここまでやる必要ある？」と懐疑的だった。でも変換ロジックのバグを何度も事前に捕捉してくれて、今は手放せなくなっている。

Soda Coreでソースデータの入口を守る

dbt testは変換後のモデルを守るが、そもそも上流からおかしいデータが来た場合を検知するには別のレイヤーが必要だ。うちではSoda Coreを使って、生データが着地したS3バケット・Redshiftの生テーブルに対してチェックをかけている。

Soda CoreのYAML構文は直感的で、エンジニア以外のアナリストでも書けるのが地味に助かっている。実際、「自分でチェック条件を追加したい」というアナリストが自走してくれるようになった。

# checks/raw_transactions.yml
checks for raw_transactions:
  # 基本的なデータ量チェック
  - row_count > 1000:
      name: "取引データが最低1000件あること"
      warn: when between 1000 and 5000
      fail: when < 1000
  
  # NULL率の監視
  - missing_percent(amount) < 1%:
      name: "金額フィールドのNULL率1%未満"
  
  # 値の範囲チェック（外れ値検知）
  - max(amount) < 10000000:
      name: "単一取引の最大額が1000万円未満"
  
  # スキーマドリフト検知（2025年から強化）
  - schema:
      name: "スキーマが期待通りであること"
      warn:
        when schema changes:
          - column add
      fail:
        when schema changes:
          - column delete
          - column type change

実行はAirflowのDAGに組み込んでいる。

# dags/data_quality_pipeline.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'on_failure_callback': notify_slack_on_failure,  # 失敗時のSlack通知
}

with DAG(
    'data_quality_pipeline',
    default_args=default_args,
    schedule_interval='0 */6 * * *',  # 6時間ごと
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    # 1. ソースデータ品質チェック
    soda_check_source = BashOperator(
        task_id='soda_check_source',
        bash_command='''
            soda scan -d redshift_prod -c configs/soda_config.yml \
                checks/raw_transactions.yml
        ''',
    )

    # 2. dbt変換実行
    dbt_run = BashOperator(
        task_id='dbt_run',
        bash_command='dbt run --select finance --target prod',
    )

    # 3. dbt test実行
    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='dbt test --select finance --target prod --store-failures',
    )

    # 4. 変換後の品質チェック
    soda_check_mart = BashOperator(
        task_id='soda_check_mart',
        bash_command='''
            soda scan -d redshift_prod -c configs/soda_config.yml \
                checks/fct_sales.yml
        ''',
    )

    soda_check_source >> dbt_run >> dbt_test >> soda_check_mart

データパイプラインで2年間ハマり続けた話でも触れているが、このチェックの実行順序設計が結構重要で、「ソース → 変換 → マート」の各ステージでフェイルファストする構造にすることで、問題のトレースが劇的に楽になった。

Monte Carloで「異常の自動学習」を任せる

dbt testとSoda Coreでルールベースのチェックは固められた。でも実務で困るのは「ルールを書けない異常」だ。

例えば、毎週月曜日は売上が高めで週末は低め、という季節性のあるデータに対して「1日の売上が○円以下なら異常」とは書きにくい。月曜と日曜で正常範囲が全然違うから。ここで統計的な異常検知が必要になる。

2026年現在、Monte Carloがこの領域では一番成熟していると感じている。特徴は、過去のデータパターンから自動的にベースラインを学習して、そのパターンから外れた場合だけアラートを出してくれること。たとえばこんなイメージだ。

xychart-beta
    title "Monte Carlo異常検知：日別売上と信頼区間"
    x-axis [月, 火, 水, 木, 金, 土, 日]
    y-axis "売上（万円）" 0 --> 1000
    bar [820, 750, 780, 810, 890, 420, 310]
    line [800, 740, 760, 800, 870, 410, 300]

うちでは以下のような監視項目を設定している。

監視項目	検知タイプ	アラート閾値
テーブル行数の変化	統計的異常	90日移動平均±3σ
NULL率の急変	絶対値 + 傾向変化	前日比+5%以上
カラム値分布の変化	分布比較	KSテストp値 < 0.05
スキーマ変更	イベント検知	変更即時通知
フレッシュネス	時刻ベース	期待更新時刻+1時間

Monte Carloのデメリットを正直に言うと、コストが高い。チームの規模や監視対象テーブル数によるが、中規模チームでも月数十万円になることがある。正直まだ費用対効果を検証中で、OSS代替としてMetricflowやElementaryも並行で試している。

実際の運用で気づいた品質スコアの可視化

仕組みを作っただけでは続かない。データ品質を「チームが意識し続ける文化」にするために、品質スコアのダッシュボードを作った。

pie title データ品質テスト結果分布（2026年7月）
    "PASS" : 847
    "WARN" : 43
    "FAIL" : 12
    "SKIP" : 8

このダッシュボードをWeeklyの定例ミーティングで毎回確認するようにしたことで、データ品質が「エンジニアだけの問題」じゃなくてチーム全体の関心事になってきた。アナリストも「このテーブルのFAILって何が起きてる？」と聞いてくれるようになって、これは地味に大きな変化だった。

品質スコアの計算式はシンプルにしている。

# scripts/quality_score.py
import boto3
from datetime import datetime, timedelta

def calculate_quality_score(table_name: str, days: int = 7) -> dict:
    """
    データ品質スコアを計算する
    スコア = PASS数 / (PASS数 + FAIL数) * 100
    """
    # dbt test結果をS3から取得
    s3 = boto3.client('s3')
    results = fetch_test_results(table_name, days)
    
    total = len(results)
    passed = sum(1 for r in results if r['status'] == 'pass')
    failed = sum(1 for r in results if r['status'] == 'fail')
    warned = sum(1 for r in results if r['status'] == 'warn')
    
    score = (passed / total * 100) if total > 0 else 0
    
    # 重大度別の重み付きスコアも計算
    weighted_score = (
        passed * 1.0 +
        warned * 0.7 +  # WARNは70%のポイント
        failed * 0.0
    ) / total * 100 if total > 0 else 0
    
    return {
        'table': table_name,
        'score': round(score, 1),
        'weighted_score': round(weighted_score, 1),
        'total_tests': total,
        'passed': passed,
        'failed': failed,
        'warned': warned,
        'calculated_at': datetime.utcnow().isoformat(),
    }

def notify_quality_degradation(table_name: str, score: float, threshold: float = 95.0):
    """品質スコアが閾値を下回ったらSlackに通知"""
    if score < threshold:
        message = (
            f"🚨 データ品質アラート\n"
            f"テーブル: `{table_name}`\n"
            f"品質スコア: {score}% (閾値: {threshold}%)\n"
            f"対応が必要です。Grafanaダッシュボードで詳細確認してください。"
        )
        send_slack_alert(message, channel='#data-quality-alerts')

ここは好みが分かれるかもしれないけど、うちでは「95%以上を合格ライン」として設定している。100%を目指すより「何が95%を下げているか」を継続的に改善する方が現実的だという判断だ。

品質スコアの推移を見ると、導入前後で明確な差が出た。

xychart-beta
    title "データ品質スコア推移（%）"
    x-axis ["1月", "2月", "3月", "4月", "5月", "6月", "7月"]
    y-axis "品質スコア(%)" 60 --> 100
    line [68, 71, 75, 88, 92, 95, 97]

1〜2月のスコアが低いのは、品質が悪かったというより、チェックを入れ始めて初めて「今まで見えていなかった問題」が可視化されたから。逆に言えば、あの数字は「以前からずっとこうだった」という現実だ。スコアが右肩上がりなのは品質が自然に良くなったわけではなく、問題を一個ずつ修正し続けた結果だ。

チームで運用して1年、正直に振り返る

導入から約1年経って見えてきたことをまとめる。

良かったこと

まず、データへの信頼性が上がった。以前は「このデータ、本当に正しい？」という会話がダッシュボードを見るたびに起きていたが、今は品質スコアが可視化されているので「今週95%だから大丈夫」という共通認識で話せる。

問題の検知が劇的に早くなった。以前は「ダッシュボードがおかしい」というユーザー報告で問題を知っていたが、今はパイプライン実行直後に検知できる。SLO設計で2年失敗し続けた僕が、ようやく運用が回り始めた話と同じで、「測定できることしか改善できない」というのがデータ品質でも痛感できた。

しんどかったこと

テストの「メンテナンスコスト」は最初に思っていたより重い。スキーマが変わるたびにテスト定義を更新しないといけない。特にdbt testはモデルファイルと同期を取り続ける必要があって、開発速度が上がると追いつかなくなることがある。個人的にここが一番しんどかった。

アラートの「疲労」も問題になった。最初は全部Slackに流していたら、1日に何十件も飛んでくるようになって誰も見なくなった。全部通知すると全部無視される、という当然すぎる結末。今は重大度でフィルタリングして、FAILのみを即時通知、WARNは日次サマリーにまとめる設計に変えている。

正直まだ検証中なのが「データドリフト」の検知だ。スキーマ変更は検知できるが、「意味的な変化」——例えばユーザーIDの採番方式が変わったとか、そういうレベルの変化はルールベースでは難しい。2026年現在、LLMを使ったデータ変更の意味理解は実験段階で、実用レベルには達していないと感じている。

データ品質管理と切り離せない話として、データカタログの整備も同時に必要になってくる。テーブルの定義が整備されていないと「何が正しい値か」がそもそも定義できないから。データカタログ3回失敗して分かった、ようやく「使われる」ようになった話も参考にしてほしい。

まとめ

1年間データ品質管理の仕組みを整えてきて、要点を整理するとこうなる。

• 変換レイヤー（dbt test）とソースレイヤー（Soda Core）の両方でチェックする — どちらか一方では盲点が生まれる
• ルールベース（閾値チェック）と統計ベース（Monte Carlo的な異常検知）を組み合わせる — 書けるルールは明示的に、書けない異常はモデルに任せる
• 品質スコアを可視化してチームの共通言語にする — エンジニアだけが見るメトリクスにしない
• アラートは「疲れない量」に絞る — 全部通知すると全部無視される
• テストのメンテナンスコストを見積もりに入れる — 導入よりも継続運用の方がしんどい

次のアクションとしては、まずdbt testから始めるのが一番取っ掛かりやすいと思う。既存のdbtプロジェクトにdata_testsを追加するだけで、最低限の品質ゲートが作れる。Soda CoreはYAMLを書くだけで動くので2週間あれば導入できる。Monte Carloのような統計的異常検知は、ルールベースで対応しきれない問題が出てきてから検討しても遅くない。

皆さんのチームでは今、データ品質のチェックってどうやってますか？ルールベースで全部カバーしようとして泥沼にハマっている人がいたら、ぜひ経験を聞かせてほしい。