データエンジニアリング

売上データが吹っ飛んだ夜から9ヶ月、データ品質管理を定着させるまでの話

深夜に売上150%急騰、原因は重複インサート——そんなインシデントを経て本気でデータ品質管理に向き合った記録。dbt・GE・Monte Carloを実際に使って見えた「ツール選定の正解と後悔」を正直に書きます。

データエンジニアリングデータ品質管理dbtGreat Expectationsデータパイプライン

深夜2時、売上データが150%跳ね上がった

去年の秋のことだ。夜中に「売上が昨日比150%になってる、なんか変」というSlackが飛んできた。最初はキャンペーンの効果かと思ったんだけど、ソースデータを追いかけたら重複インサートが混入していた。ダウンストリームのBIダッシュボードに乗り込む前にアラートが来たのは運が良かったが、あのインシデントがうちのチームにデータ品質管理を本気で導入するきっかけになった。

あの夜の詳細は本番で売上データが150%跳ね上がった日、データ品質管理と向き合った話に書いたのでそちらも読んでほしい。今回はその後、約9ヶ月かけてチームに定着させた実装方法を、失敗も含めて正直に書く。

2026年のデータ品質ツール、実際に使って比較した結果

まず最初に、ツール選定で2ヶ月かけて迷走した話をしよう。候補は大きく5つあった。

ツール特徴料金体系学習コストうちの評価
Great Expectations 1.xOSSの定番。ruleベース無料 (GX Cloud有料)中〜高★★★★☆
dbt Tests + dbt-expectationsdbtに統合できるOSS無料低★★★★★
Monte Carloカタログ+異常検知従量課金 (高め)低★★★☆☆
Soda CoreGE代替、YAMLベース無料+Soda Cloud中★★★☆☆
AWS Glue Data QualityAWSネイティブGlue DPU低★★★☆☆

結論から言うと、うちはdbt Tests + dbt-expectationsをメインに据えて、異常検知の部分だけAWS Glue Data Qualityを組み合わせた構成に落ち着いた。Great Expectationsは一時期使っていたんだけど、1.x系でAPIが大幅に変わってドキュメントが追いつかず、チームの学習コストが想定の倍かかった。正直まだ1.xは成熟しきっていないと思う。

Monte Carloは機能自体はすごくて、デモを見たときは「これだ！」と思ったんだけど、価格が月50万円を超えて予算会議を通過できなかった（笑）。いいものが高いのは仕方ないとはいえ、さすがに厳しい。

パイプライン全体のアーキテクチャ

品質検証をどこに挟むかで悩んだ結果、こういう構成になった。

flowchart TB
    subgraph Sources["データソース"]
        A["RDS (トランザクション DB)"]
        B["S3 Raw Data"]
        C["Kinesis Stream"]
    end

    subgraph Ingestion["取り込み層"]
        D["AWS Glue 5.0"]
        E["Lambda ETL"]
    end

    subgraph Staging["Staging層"]
        F["Redshift Serverless"]
        G["S3 Iceberg"]
    end

    subgraph QualityLayer["品質検証層 ★"]
        H["dbt Tests"]
        I["AWS Glue Data Quality"]
        J["Custom Anomaly Detector"]
    end

    subgraph Alert["アラート・可視化"]
        K["CloudWatch Alarm"]
        L["Slack 通知"]
        M["Grafana Dashboard"]
    end

    subgraph Serving["提供層"]
        N["BIダッシュボード"]
        O["API エンドポイント"]
    end

    A --> D
    B --> D
    C --> E
    D --> F
    D --> G
    E --> F
    F --> H
    G --> I
    H --> J
    I --> J
    J --> K
    K --> L
    K --> M
    H --> N
    H --> O

品質検証層がパイプライン中央に位置しているのがポイントだ。検証を通過しないデータはServingには絶対に出ない、という設計にした。「品質チェックは最後にやればいい」と思いがちだけど、それだとダウンストリームへの汚染が防げない。この構成にしてから「気づいたら変なデータがBIに乗ってた」という事態がなくなった。

dbt Testsで実際に書いた品質ルール

dbt Tests + dbt-expectationsの組み合わせは個人的にマジで便利で、「なんでもっと早く入れなかったんだ」という気持ちになった。YAMLとSQLで書けるので、データエンジニア以外のメンバーでも読めるのが地味に大きい。実際に本番で動いているコードを一部公開する。

schema.yml（基本的なテスト）

# models/marts/sales/schema.yml
version: 2

models:
  - name: fct_orders
    description: "受注ファクトテーブル"
    columns:
      - name: order_id
        tests:
          - unique
          - not_null

      - name: order_amount
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 10000000  # 1000万円を超える注文はデータ異常とみなす

      - name: created_at
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_of_type:
              column_type: timestamp

      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'processing', 'shipped', 'delivered', 'cancelled']

    tests:
      # テーブルレベルのテスト
      - dbt_expectations.expect_table_row_count_to_be_between:
          min_value: 1000   # 1日の注文が1000件を下回ったら異常
          max_value: 1000000
      - dbt_expectations.expect_table_columns_to_match_ordered_list:
          column_list:
            - order_id
            - customer_id
            - order_amount
            - status
            - created_at

カスタムテスト（重複インサート検知）

あの夜のインシデントの原因だった重複インサートは、カスタムテストを書いて防ぐようにした。

-- tests/generic/no_duplicate_within_window.sql
{% test no_duplicate_within_window(model, column_name, partition_by, window_hours=1) %}

WITH duplicates AS (
    SELECT
        {{ column_name }},
        COUNT(*) AS cnt,
        DATE_TRUNC('hour', created_at) AS window_start
    FROM {{ model }}
    WHERE created_at >= DATEADD('hour', -{{ window_hours }}, CURRENT_TIMESTAMP)
    GROUP BY {{ column_name }}, DATE_TRUNC('hour', created_at)
    HAVING COUNT(*) > 1
)

SELECT * FROM duplicates

{% endtest %}
# schema.yml に追加
- name: order_id
  tests:
    - no_duplicate_within_window:
        partition_by: customer_id
        window_hours: 2

このカスタムテストのおかげで、あの夜のようなインシデントは再発していない。派手さはゼロだけど、本当に効いてる。

dbt run-operation で実際に動かしてみた結果

$ dbt test --models fct_orders --select test_type:generic

08:15:23  Running with dbt=2.0.1
08:15:24  Found 47 models, 183 tests, 0 sources, 0 exposures
08:15:25  
08:15:25  Concurrency: 4 threads (target='prod')
08:15:25  
08:15:26  1 of 183 START test unique_fct_orders_order_id ........................ [RUN]
08:15:28  1 of 183 PASS unique_fct_orders_order_id .............................. [PASS in 1.82s]
08:15:28  2 of 183 START test not_null_fct_orders_order_amount .................. [RUN]
08:15:29  2 of 183 PASS not_null_fct_orders_order_amount ........................ [PASS in 0.94s]
...
08:19:41  181 of 183 PASS expect_table_row_count_to_be_between_fct_orders ...... [PASS in 2.11s]
08:19:43  182 of 183 FAIL no_duplicate_within_window_fct_orders_order_id ....... [FAIL 3 in 1.55s]
08:19:44  183 of 183 PASS expect_column_values_to_be_between_order_amount ....... [PASS in 0.89s]

08:19:44  Finished running 183 tests in 0 minutes and 4 minutes 21.32 seconds.

08:19:44  Completed with 1 failures:
08:19:44    - Failure in test no_duplicate_within_window_fct_orders_order_id (tests/generic/no_duplicate_within_window.sql)
08:19:44      Got 3 results, configured to fail if != 0

こういう感じで失敗した時にすぐ分かるのがいい。CIに組み込んでいるのでプルリク段階で検知できる。出力がシンプルで人間が読みやすいのも、地味に運用コストを下げてくれている。

異常検知の仕組みを自前で実装した話

dbt Testsでルールベースの検証はできるんだけど、「統計的な異常」を検知するのは別の仕組みが必要だった。Monte Carloのような商用ツールが自動でやってくれる部分だ。

予算の制約から自前実装を選んだんだけど、これが思ったより良いものができた。

# anomaly_detector.py
import boto3
import pandas as pd
import numpy as np
from scipy import stats
from datetime import datetime, timedelta
import json

class DataQualityAnomalyDetector:
    def __init__(self, redshift_client, cloudwatch_client):
        self.redshift = redshift_client
        self.cloudwatch = cloudwatch_client
        self.lookback_days = 30  # 過去30日の統計を使う

    def detect_row_count_anomaly(self, table_name: str, today_count: int) -> dict:
        """
        過去30日の行数分布から z-score で異常を検知
        z-score > 3 なら異常と判断 (99.7%信頼区間)
        """
        historical_counts = self._get_historical_row_counts(table_name)
        
        if len(historical_counts) < 7:  # 統計的に意味があるデータが揃うまでスキップ
            return {"status": "insufficient_data", "table": table_name}
        
        mean = np.mean(historical_counts)
        std = np.std(historical_counts)
        
        if std == 0:
            return {"status": "constant", "table": table_name}
        
        z_score = abs((today_count - mean) / std)
        
        result = {
            "table": table_name,
            "today_count": today_count,
            "historical_mean": round(mean, 2),
            "historical_std": round(std, 2),
            "z_score": round(z_score, 3),
            "is_anomaly": z_score > 3.0,
            "severity": self._get_severity(z_score)
        }
        
        # CloudWatch にメトリクスを送る
        self.cloudwatch.put_metric_data(
            Namespace='DataQuality',
            MetricData=[
                {
                    'MetricName': 'RowCountZScore',
                    'Dimensions': [{'Name': 'TableName', 'Value': table_name}],
                    'Value': z_score,
                    'Unit': 'None',
                    'Timestamp': datetime.utcnow()
                }
            ]
        )
        
        return result

    def _get_severity(self, z_score: float) -> str:
        if z_score < 2.0:
            return "normal"
        elif z_score < 3.0:
            return "warning"
        elif z_score < 5.0:
            return "critical"
        else:
            return "emergency"

    def _get_historical_row_counts(self, table_name: str) -> list:
        # Redshiftの履歴テーブルから行数を取得
        # 実際はAWS Redshiftのクエリで取得
        # ここでは省略
        pass


# Lambda で毎日実行
def lambda_handler(event, context):
    detector = DataQualityAnomalyDetector(
        redshift_client=boto3.client('redshift-data'),
        cloudwatch_client=boto3.client('cloudwatch')
    )
    
    tables_to_monitor = [
        'fct_orders', 'fct_revenue', 'dim_customers', 'dim_products'
    ]
    
    results = []
    for table in tables_to_monitor:
        today_count = get_today_row_count(table)  # 省略
        result = detector.detect_row_count_anomaly(table, today_count)
        results.append(result)
        
        if result.get('is_anomaly'):
            send_slack_alert(result)  # 省略
    
    return {"statusCode": 200, "results": results}

ただ正直まだ改善の余地はある。「曜日の周期性」（月曜は売上が少ない、など）を考慮していないので、月曜日の行数が少ないと誤検知することがある。今後 Prophet あたりを使って時系列モデルに変えようと思っているが、そこまで手が回っていないのが現状だ。

9ヶ月運用して見えたデータ品質スコアの変化

導入前後でどれだけ改善したか、実際の数値で見せよう。

xychart-beta
    title "データ品質スコアの推移（月次）"
    x-axis ["2025-10", "2025-11", "2025-12", "2026-01", "2026-02", "2026-03", "2026-04", "2026-05", "2026-06"]
    y-axis "品質スコア (0-100)" 0 --> 100
    line [52, 58, 67, 74, 78, 83, 86, 89, 91]
    bar [52, 58, 67, 74, 78, 83, 86, 89, 91]

スコアは dbt Testsのパス率・異常検知の精度・インシデント件数を組み合わせて算出した独自指標だ。導入直後の2025年10月が52点だったのが、2026年6月には91点まで改善している。

特に3ヶ月目から急激に改善しているのは、「アラートが多すぎて誰も見なくなる問題」を解決したからだ。これはデータ品質管理の導入で一番ありがちな失敗で、SLO設計で2年間失敗し続けた僕が、ようやく運用が回り始めた話でも似たことを書いた。アラートの閾値を厳しくしすぎると人が疲弊する。ツールより先に、人間が続けられる運用設計を考えるべきだった。

インシデント件数の変化

xychart-beta
    title "データ品質起因のインシデント件数（月次）"
    x-axis ["2025-10", "2025-11", "2025-12", "2026-01", "2026-02", "2026-03", "2026-04", "2026-05", "2026-06"]
    y-axis "件数" 0 --> 20
    bar [18, 15, 11, 8, 6, 4, 3, 2, 1]

月18件あったインシデントが月1件になった。しかもその1件も「事前に検知してダウンストリームへの影響なし」という形で処理できている。数字で見るとわかりやすいけど、チームの心理的な余裕がまったく変わった、というのが個人的には一番大きかった。

チームに定着するまでに犯した3つの失敗

ツールを入れるだけでは定着しないんですよね。これが一番痛かった部分なので正直に書く。

失敗1: テストを書きすぎた

最初の2ヶ月で300個以上のテストを書いた。結果、実行時間が45分を超えてCIがボトルネックになった。今は重要なテーブルだけに絞って150個程度にしている。「全部カバーしたい」気持ちはわかるんだけど、実行時間10分以内に抑えないとチームが回らないと感じた。網羅性より継続性、という話だ。

失敗2: 通知先をチャンネル全体にした

最初はSlackのチャンネル全体に通知を飛ばしていた。アラートが1日20件以上来るようになって、みんなが無視するようになった。典型的な「オオカミが来た」状態だ。今は severity ごとに通知先を変えている。

flowchart LR
    A["品質チェック失敗"] --> B{severity判定}
    B --> |"emergency"| C["📢 全員メンション + PagerDuty"]
    B --> |"critical"| D["#data-quality-alert チャンネル"]
    B --> |"warning"| E["#data-quality-log チャンネル (通知なし)"]
    B --> |"normal"| F["ログのみ、通知なし"]

失敗3: データオーナーを決めなかった

テーブルごとの担当者を決めず「みんなの問題」にしたら、誰も直さなかった。「誰でも直せる」は「誰も直さない」と同義だ。今はdbtのschema.ymlに meta.owner フィールドを必須にして、アラートも担当者に直接飛ぶようにした。

models:
  - name: fct_orders
    meta:
      owner: "@tanaka"
      team: "data-platform"
      sla: "08:00 JST"

データカタログとの連携についてはデータカタログ完全ガイド2026｜ツール比較・AI活用・導入設計も参考になると思う。

2026年の最新動向：AIを使った品質チェック

2026年に入ってから試し始めた取り組みとして、LLMを使った「意味的な異常検知」がある。ルールベースでは検知できない「値は正しいけど意味がおかしい」パターンを拾えないか実験中だ。

# 実験中のコード（まだ本番未適用）
import anthropic

def semantic_quality_check(table_sample: dict, table_description: str) -> dict:
    """
    LLMにサンプルデータを見せて意味的な異常を検知する
    例:「顧客IDが000000000のレコードが大量にある」など
    """
    client = anthropic.Anthropic()
    
    prompt = f"""
    以下のテーブルのサンプルデータを確認してください。
    テーブルの説明: {table_description}
    
    サンプルデータ:
    {table_sample}
    
    データ品質の観点から、以下を判断してください:
    1. 意味的に不自然な値はあるか
    2. 不自然だと判断した根拠
    3. severity (normal/warning/critical)
    
    JSON形式で回答してください。
    """
    
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return json.loads(response.content[0].text)

まだ実験段階で、コストと精度のバランスが取れていない。全テーブルに適用するのは現実的じゃないので、「重要テーブルの日次サンプルチェック」程度の使い方を検討している。正直まだ検証中だけど、ルールで書き切れない「なんとなくおかしい」を拾える可能性があって、面白い方向性だと思っている。

まとめ

9ヶ月のデータ品質管理導入から得た主な知見を整理する。

  1. ツール選定は「チームが続けられるか」で選ぶ — 高機能な商用ツールより、dbtに統合できるシンプルなものの方がチームに定着した。学習コストを甘く見ると後で痛い目を見る。

  2. テストの数より「重要なテーブルをちゃんと守れているか」を優先する — 300個のテストより、コアな20テーブルの150個の方がずっと価値がある。CIの実行時間10分以内を死守すること。

  3. アラート設計で運用が決まる — severityごとに通知先を分け、emergency以外は自動でログに流す設計にしないと、全員がアラートを無視するようになる。

  4. データオーナーを明確に決める — テーブルごとの担当者をコード（schema.yml）に書くことで、責任の所在が明確になる。

  5. AIによる意味的チェックは面白いが、まだ補完的な位置付け — ルールベースの検証が揃ってから試す順序が正しい。

次のアクション: まずdbt Testsを1テーブルだけに入れてみることをお勧めする。uniqueとnot_nullだけでも、意外と問題が見つかるはずだ。完璧なシステムを一気に作ろうとすると確実に挫折する。小さく始めて、インシデントのたびにテストを追加するサイクルが一番続く。

チームのデータ品質管理で困っていることがあれば、コメントかTwitterで気軽に聞いてほしい。

U

Untanbaby

ソフトウェアエンジニア｜AWS / クラウドアーキテクチャ / DevOps

10年以上のIT実務経験をもとに、現場で使える技術情報を発信しています。 記事の誤りや改善点があればお問い合わせからお気軽にご連絡ください。

関連記事

データエンジニアリング

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

データエンジニアリング

本番で売上データが150%跳ね上がった日、データ品質管理と向き合った話

データエンジニアリング

売上データが消えた夜から9ヶ月——データ品質管理を「使われる仕組み」に変えた実装記

データエンジニアリング

dbt Core 2.0移行後1年、本番運用で積み上がった設計の知見

データエンジニアリング

データパイプラインで2年間ハマり続けた話｜Airflow・dbt・Kafka本番運用の現実

データエンジニアリング

A/Bテスト基盤を自前構築して1年半、失敗から学んだデータエンジニアリングの現実

← 記事一覧に戻る