データエンジニアリング

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

朝のダッシュボードで売上が150%跳ね上がる事態に。2時間で気づいた重複計上バグから始まった、実践的なデータ品質管理の構築記。dbt・Great Expectations・アノマリー検出まで、本当に使える検証ルール設計を公開します。

データエンジニアリングデータ品質dbtGreat ExpectationsApache Spark

売上データが1時間で消えた夜から始まった

先日、朝7時に緊急Slackが鳴った。「売上データが150%跳ね上がってる」。リアルタイムダッシュボードを開いたら、昨日の売上が30万円だったのに、今朝は74万円。営業部は喜んでるけど、僕は頭真っ白だった。

原因は単純だった。ETLパイプラインのある段階で、デューバリケーション処理が外れていたんだ。同じトランザクションが3回重複で計上されていた。それを発見するのに2時間かかった。その2時間の間に、営業がすでにその数字を経営層に報告していた。

「これはマズい」って思って、チームに掛け合ったのが、本格的にデータ品質管理に向き合うきっかけだった。以来9ヶ月、検証ルール設計からアノマリー検出まで、地獄のような試行錯誤を重ねた。今、その実装がようやく「使われる状態」になった。

最初は dbt test だけで充分だと思ってた

最初の判断は間違ってなかった。dbtでカラムのNULLチェックと基本的な統計値の検証を入れたんだ。not_null、unique、relationshipsといった基本テストに加えて、カスタムクエリで「前日との差分が-50%を超えないか」みたいなロジックも用意した。

models:
  - name: daily_sales
    columns:
      - name: order_id
        tests:
          - not_null
          - unique
      - name: amount
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: "amount >= 0"
      - name: created_at
        tests:
          - relationships:
              to: ref('users')
              field: id
    tests:
      - dbt_utils.equal_rowcount:
          compare_model: source('raw', 'transactions')

これで基本的な検証は回ってた。でも問題があったんだ。テストは夜間バッチ中に走るから、朝には既に壊れたデータがダッシュボードに載ってる んですよね。テストで異常を検出しても、その時点で既にビジネスロジックが動き終わってた。つまり、検出は遅すぎたわけだ。

Great Expectations で「パイプライン中」の検証を実装した

そこで Great Expectations を導入した。dbt testとの違いは、パイプライン実行中にデータの形状をリアルタイムで チェック できる点だ。

PySpark でロード直後、変換前に統計量を計算して、過去のプロファイルと比較する。「このカラムの平均値が±30%ズレてたらアラート」みたいなチェックをdata-docsで視覚化できるのが地味に便利だった。

import great_expectations as ge
from great_expectations.core.batch import RuntimeBatchRequest

context = ge.get_context()

# データロード直後
df = spark.sql("SELECT * FROM transactions WHERE date = CURRENT_DATE")
validation_df = ge.from_pandas(df.toPandas())

# 期待値定義
expectation_suite = context.create_expectation_suite(
    expectation_suite_name="transactions_daily",
    overwrite_existing=True
)

validation_df.expect_column_values_to_be_in_set(
    column="status",
    value_set=["completed", "pending", "failed"]
)

validation_df.expect_column_mean_to_be_between(
    column="amount",
    min_value=1000,
    max_value=50000
)

validation_df.expect_column_value_lengths_to_be_between(
    column="order_id",
    min_value=36,
    max_value=36
)

result = context.run_validation(
    validator=validation_df
)

これで実行時に異常を検出できるようになった。ただし、すぐに次の問題が出てきた。どのルールを信じればいいのか ってやつだ。

最初は「とにかく厳しくチェック」してたから、毎日20〜30件のアラートが出た。営業部はイライラしたし、データエンジニアリングチームは対応で手いっぱいになった。結局、アラート疲れで何も対応しなくなったんだ。あるあるですね。

ドリフト検出で「本当に異常か」を機械学習で判定

そこで気づいたのは、「異常」って定義が固定値じゃダメだ ってこと。業界の季節性、キャンペーン期間、新規顧客の流入パターン……こういう文脈的な情報を入れないと、単なるノイズアラートになるんですよ。

そこで Whylogs という異常検出ツールに切り替えた。統計量の時系列変化を追跡して、「いつから外れてるのか」を明確にする。

import whylogs as why
from whylogs.core.profile import SegmentedDataFrame

# 日々のプロファイル記録
profile = why.log(df).profile()

# ドリフト検出
baseline_profile = why.read("s3://baseline/2026-08-01.profile")
drift_report = baseline_profile.get_drift_report(profile)

# 実際には Evidently も試した
from evidently.report import Report
from evidently.metric_preset import DataQualityPreset

report = Report(metrics=[
    DataQualityPreset()
])

report.run(
    reference_data=df_baseline,
    current_data=df_today
)

print(report.as_dict())

これでやっと「本当に異常か、ただのノイズか」が区別できた。例えば、order amountの平均が12,000円から15,000円に上がった。でもWhylogsが「キャンペーンテストの効果では？」と教えてくれたんだ。その情報をビジネス側に共有すると、むしろ喜ばれた。データエンジニアが単に「アラート出しました」じゃなくて「この変化の理由はおそらくこれですよ」って仮説を提示できるようになった。

アラート設計——最初はメール、今は Slack + Datadog

検証ルールの次は、アラートの行き先 問題だった。実装してから気づいたけど、ここが意外と重要だ。

最初はメール。すぐ誰にも見られなくなった。Slackに変更したけど、1日500件のメッセージが流れるから、データ品質アラートが埋もれる。そこでSlackのthread + カスタムemojiでpriorityを分け、Datadogに統合してインシデント自動作成できるようにした。

from datadog_api_client.v1.api.events_api import EventsApi
from datadog_api_client.v1.model.event_create_request import EventCreateRequest
from datadog_api_client.v1.model.event_priority import EventPriority

api_client = EventsApi()

if severity == "critical":
    priority = EventPriority.NORMAL
    message = f"🔴 CRITICAL: {table_name} - {rule_name} failed"
    tags = ["data-quality", "incident", "critical"]
    create_incident = True
elif severity == "warning":
    priority = EventPriority.LOW
    message = f"🟡 WARNING: {table_name} - possible anomaly detected"
    tags = ["data-quality", "warning"]
    create_incident = False

event = EventCreateRequest(
    title=f"Data Quality Alert: {table_name}",
    text=message,
    priority=priority,
    tags=tags,
    alert_type="info"
)

api_client.create_event(body=event)

重要なポイントは、alerting thresholdを営業・分析チームと明示的に合意する ことだ。「このメトリクスが±20%変わったら即座に知りたい」とか「このカラムがNULLになったら絶対教えて」とか、そういう明確なSLAを決める。最初はデータエンジニアが一人で決めてたけど、それは失敗だった。

正直に言うと、今でも完璧じゃない

9ヶ月運用して、月の異常発見数は50件前後。そのうち実際の問題は30件、残りはfalse positive。False positiveの率が高いのは、正直に言って設定の粗さだ。

例えば、毎週金曜日の夜は決済が集中するから、金曜と他の曜日では平均が違う。それをseasonal decompositionで自動抽出できればいいんだけど、手でセグメント分けしてる。月曜〜木曜と金曜〜日曜で別々のベースラインを持つ。めんどくさい。

でも、この「めんどくささ」を自動化しようとして、ARIMAとかProphetとか試したけど、結局オーバーフィットして余計にfalse positiveが増えた。

# 試した Prophet。結局捨てた
from fbprophet import Prophet

m = Prophet(
    yearly_seasonality=True,
    weekly_seasonality=True,
    daily_seasonality=False,
    interval_width=0.95
)

m.fit(df)
future = m.make_future_dataframe(periods=7)
forecast = m.predict(future)

# でも小さいデータセットでは過学習するし、
# 外れ値に弱いし、結局カスタムなルール書き直した

今は、「データエンジニアが1週間に1回、異常検出の結果を見直して、ルールを手で調整する」という運用になってる。自動化じゃなくて「半自動化」だ。でも、それが正解だったんですよね。完璧な自動化を目指すより、人間が関与する仕組みのほうが、実際には信頼性が高いし、チーム全体の学習にもなる。

dbt × Great Expectations × Datadog の現在形

現在のパイプラインはこんな構成になってる。

flowchart TB
    subgraph "Data Source"
        SRC["Raw Transactions<br/>S3/Database"]
    end
    
    subgraph "ETL Pipeline"
        LOAD["Load Data<br/>PySpark"]
        VALIDATE1["Great Expectations<br/>Statistical Validation"]
        TRANSFORM["dbt Transform<br/>SQL"]
        VALIDATE2["dbt Tests<br/>Structural Tests"]
    end
    
    subgraph "Quality Monitoring"
        WHYLOGS["Whylogs Profiling<br/>Drift Detection"]
        PROFILE["Historical Profiles<br/>S3"]
    end
    
    subgraph "Alerting"
        DATADOG["Datadog Events<br/>Anomaly Aggregation"]
        SLACK["Slack Notifications<br/>Incident Routing"]
    end
    
    subgraph "Output"
        MART["Mart Tables<br/>BI Ready"]
        DASHBOARD["Business Dashboard<br/>Validated Data"]
    end
    
    SRC --> LOAD
    LOAD --> VALIDATE1
    VALIDATE1 --> TRANSFORM
    TRANSFORM --> VALIDATE2
    
    VALIDATE1 --> WHYLOGS
    VALIDATE2 --> WHYLOGS
    WHYLOGS --> PROFILE
    WHYLOGS --> DATADOG
    
    DATADOG --> SLACK
    DATADOG --> |"Critical Incidents"| SLACK
    
    VALIDATE2 --> |"Pass Only"| MART
    MART --> DASHBOARD

各段階で役割が異なるんだ：

  1. dbt: transformation直後に軽量な構造的テスト（NULL、unique、foreign key）
  2. Great Expectations: ETLロード直後に統計的テスト（分布、カテゴリ値の妥当性）
  3. Whylogs: 日々生成されたプロファイルをベースラインと比較してドリフト検出
  4. Datadog: 全てのアラートを集約して、severity別にチームに通知
  5. Slack: incident channelでcritical alertsをthread化

チームで「使われる」ようになるまで

重要なのは、これを営業・分析チームが 主体的に使う ようになったこと。

最初は「データエンジニアが検証しといたから」って他人事だった。でも、月2回の定例会議で「先週のアラート50件、実問題は30件」「false positiveの傾向を見ると〇〇パターン」って共有してたら、分析チームが自分たちで新しいcheckを提案するようになったんだ。

例えば、こんな感じで改善要望が出てくるようになった：

  • 「新規ユーザーと既存ユーザーでセグメント分けして別々にアラート出せ」
  • 「このカラムは水曜日だけNULLが多いから、曜日別のthresholdが必要」
  • 「営業プロモーション期間は事前に教えてくれたら、その日のベースラインを一時的に広げる」

こういう biz-drivenな改善 が出てくるようになった。これが「使われる」状態だと思う。最初は「データエンジニアの責任範囲」だったのが、今は「全体で品質を守る」という意識に変わった。

まとめ

データ品質管理は、検証ツール選びが30%、アラート設計が30%、チームとの合意が40%だった。 最初は技術的な完璧さを求めてたけど、実際には「ビジネスチームが信頼して使い続ける仕組み」が全てなんですよ。

  • dbt testは構造的な検証の第一線。毎日走らせて、異常は朝には必ず見つかるようにしておく
  • Great Expectationsでリアルタイム検証。ロード直後なら、下流への影響を最小化できる
  • Whylogs/Evidentlyでドリフト検出。単なる閾値超過じゃなく「本当に異常なのか」を判定する
  • アラートはSlack × Datadogでmulti-channel。criticalは即座に、warningはdata-docsで見直しレベルに
  • 営業・分析チームとmonthly review。彼らの要望を反映させることで、「他人事」から「自分事」に変わる

次は、MLパイプラインのデータ品質にも同じ仕組みを展開したい。training dataのドリフトも同じくらい重要だし、model predictionのアウトプット品質も監視する必要がある。まだ試行錯誤の段階だけど、スケーラブルな仕組みにしていく予定だ。

何か質問あったら、遠慮なく聞いてください。実装での地雷も、成功パターンも、全部シェアできます。

U

Untanbaby

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

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

関連記事

データエンジニアリング

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

データエンジニアリング

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

データエンジニアリング

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

データエンジニアリング

dbt 1.9→2.0移行で半年ハマった話｜本番で気づいたモデル設計の現実

データエンジニアリング

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

データエンジニアリング

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

← 記事一覧に戻る