2026.11.12 AI

データパイプライン監視の実装：鮮度・件数・スキーマ変化をdbtとAirflowで検知する手順【2026年版】

Q: dbtを使っていなくてもデータパイプライン監視は実装できますか？

できます。鮮度は「対象テーブルの max(_loaded_at) と現在時刻の差が閾値を超えたら1行返す」SQLを、件数は本記事のsingular testから source() を実テーブル名に置き換えたSQLを、Airflowのタスクとして定期実行すれば同じ検知ができます。結果が1行以上ならタスクを失敗させ、 on_failure_callback で通知する構成です。dbtを入れる利点は、閾値をYAMLで一元管理でき、モデルの依存関係と並べて実行できる点にあります。

データパイプライン監視の実装：鮮度・件数・スキーマ変化をdbtとAirflowで検知する手順【2026年版】

データパイプライン監視は、ジョブが落ちたことだけでなく、ジョブが成功したのにデータが古い・少ない・形が変わった状態を見つける仕組みです。本記事では、監視する信号をジョブ失敗・鮮度・件数・スキーマの4つに分け、Airflow 3.4系とdbt Core 1.12系（いずれも2026年11月時点の最新）でそれぞれを実装する手順をコード付きで示します。Slackへの通知、件数急減を検知するSQL、列定義の差分を取るSQL、通知を鳴らしすぎないための運用、専用ツールへ移る判断基準までを扱います。

まとめ：データパイプライン監視はジョブ失敗と鮮度から実装し件数・スキーマへ広げる

最初に入れるのは2つです。Airflowのon_failure_callbackでジョブ失敗をSlackへ送ること、dbtのsource freshnessで取り込みテーブルの最終更新時刻を検査すること。この2つで「止まった」「届いていない」の大半を拾えます。

次に、前週までの同曜日平均と比べる件数テストと、information_schemaの列定義の差分監視を足します。通知は警告と異常で送り先を分け、異常には直す担当者を名前で付けてください。監視対象が数十テーブルを超え、閾値の手作業調整が回らなくなった時点が、専用ツールを検討する目安です。

データパイプライン監視で検知する4つの信号と、ジョブ監視だけでは足りない理由

実装に入る前に、何を見張るのかを固定します。監視の抜けは、ほとんどが「ジョブは成功した」という報告を信じた場面で起きています。

ジョブが成功してもデータが壊れる事故の典型例と、その場で見える症状

取り込み元のAPIが空の配列を返した日、取り込みジョブは0件を書き込んで正常終了します。CSV連携で列の順番が入れ替わった日も、型が合えばロードは通る状態です。上流のバッチが3時間遅れた日は、変換ジョブが前日のデータのまま定刻に成功します。

どれもジョブの終了コードは0です。気づくのは、翌朝ダッシュボードを見た営業部長であることが多い。データパイプラインとはで整理した取込・変換・配信の各層のどこで壊れても、下流には静かに誤った数字が流れます。

ジョブ失敗・鮮度・件数・スキーマの4信号と実装する場所の対応表

本記事で実装する4信号と、それぞれをどこに書くかを先に示します。

信号	検知する事故	実装する場所	本記事の実装
ジョブ失敗・遅延	タスクの異常終了、終わらないDAG	オーケストレーター	失敗コールバック・期限通知
鮮度	上流が止まり古いデータのまま	dbt（取り込み直後）	dbt source freshness
件数	0件・半減・重複による倍増	dbt（singular test）	同曜日平均との比較SQL
スキーマ	列の追加・削除・型変更	DWHのメタデータ、dbt	列定義の差分・contract

分布（値の偏りやNULL率）やリネージを加えて「五つの柱」とする整理もあります。その考え方はデータオブザーバビリティとはで扱っており、本記事は自前で書ける4信号に絞ります。表の失敗コールバックはon_failure_callback、期限通知はDeadline Alertsです。列定義の差分にはinformation_schemaを使い、contractと組み合わせます。

検証環境を整えAirflowのジョブ失敗をSlackへ通知する実装手順

最初の信号はジョブ失敗です。Airflowの基本（DAGの書き方と環境構築）はApache Airflowとはで済ませている前提で進めます。起動とリトライの自動化はデータパイプライン自動化の実装手順で扱っています。

Airflow 3.4系・dbt Core 1.12系の検証環境と通知先

PyPIのapache-airflowでは2026年11月6日公開の3.4.0が、dbt-coreでは10月14日公開の1.12.6が最新です。DWHはPostgreSQL（18系）を例にします。通知先はSlackのIncoming Webhookで、発行したURLは環境変数に入れ、コードに直書きしません。

export SLACK_WEBHOOK_URL="発行されたWebhookのURL"
curl -X POST -H "Content-type: application/json" \
  -d '{"text": "データパイプライン監視の疎通テスト"}' \
  "$SLACK_WEBHOOK_URL"

Slackの公式ドキュメントによると、成功時はHTTP 200と本文okが返ります。投稿先チャンネルはアプリ側の設定で決まり、リクエストからは変えられません。警告用と異常用でチャンネルを分けたいなら、Webhookを2本発行します。

on_failure_callbackでタスク失敗をSlackへ通知する実装

Slack用のproviderを入れなくても、標準ライブラリだけで送れます。次のDAGは、鮮度チェックが通ったときだけdbtのビルドに進み、どのタスクが落ちてもSlackへ通知します。

import json
import os
import urllib.request

from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator


def notify_slack(text):
    req = urllib.request.Request(
        os.environ["SLACK_WEBHOOK_URL"],
        data=json.dumps({"text": text}).encode("utf-8"),
        headers={"Content-type": "application/json"},
    )
    urllib.request.urlopen(req, timeout=10)


def task_failure_alert(context):
    notify_slack(
        f"[異常] {context['task_instance_key_str']} が失敗しました"
        f"（run_id={context['run_id']}）"
    )


with DAG(
    dag_id="daily_sales_pipeline",
    schedule="0 3 * * *",
    catchup=False,
    default_args={"retries": 2, "on_failure_callback": task_failure_alert},
):
    freshness = BashOperator(
        task_id="dbt_source_freshness",
        bash_command="cd /opt/dbt && dbt source freshness",
    )
    build = BashOperator(
        task_id="dbt_build",
        bash_command="cd /opt/dbt && dbt build",
    )
    freshness >> build

default_argsに書いたので、全タスクに同じコールバックが付きます。retriesを2にしているため、1回目と2回目の失敗では呼ばれず、通知の対象はリトライを使い切った最終失敗だけです。リトライのたびに知りたい場合はon_retry_callbackを足します。

公式のCallbacks解説には注意が2つあります。CLIやUIで状態を手で変えたときはコールバックが呼ばれません。コールバック内のエラーはタスクログではなくDAG processorのログに出るため、通知が来ないときはそちらを見ます。

Deadline Alertsで終わらないDAGを検知する設定と実験的機能の扱い

失敗通知では、ハングして終わらないDAGを拾えません。Airflow 2系のSLA機能は3.0で削除され、3.1からDeadline Alertsが代わりを担っています。Deadline Alertsの公式ページの例を、キュー投入から2時間で通知する形にしたものが次のコードです。

from datetime import timedelta

from airflow.sdk import AsyncCallback, DAG, DeadlineAlert, DeadlineReference
from airflow.providers.slack.notifications.slack_webhook import SlackWebhookNotifier

with DAG(
    dag_id="daily_sales_pipeline",
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_QUEUED_AT,
        interval=timedelta(hours=2),
        callback=AsyncCallback(
            SlackWebhookNotifier,
            kwargs={"text": "[遅延] {{ dag_run.dag_id }} が期限 {{ deadline.deadline_time }} を超過"},
        ),
    ),
):
    ...

基準時刻は、キュー投入時刻（DAGRUN_QUEUED_AT）のほか、予定開始時刻、固定日時、過去の成功実行の平均時間（AVERAGE_RUNTIME）から選べます。AsyncCallbackはTriggererで動くため、Triggererのプロセスが起動していることが前提です。

公式ページは、この機能を「experimental」と明記しています。予告なく仕様が変わりうるので、業務の締め時刻に直結するDAGだけに付け、Airflowの更新時はリリースノートで変更点を確かめる運用にしてください。

dbt source freshnessでテーブルの鮮度を監視するYAMLと実行コマンド

2つ目の信号は鮮度です。dbtの環境構築はdbt Coreの使い方を前提にします。

warn_afterとerror_afterを業務の許容遅延から決めるYAML定義

取り込みテーブルに、ロード時刻を入れる_loaded_at列がある例です。models/sources.ymlに次のように書きます。

sources:
  - name: ec
    schema: raw_ec
    config:
      freshness:
        warn_after: {count: 6, period: hour}
        error_after: {count: 24, period: hour}
      loaded_at_field: _loaded_at
    tables:
      - name: orders
      - name: customers
        config:
          freshness:
            warn_after: {count: 2, period: day}
            error_after: {count: 3, period: day}
      - name: product_master
        config:
          freshness: null

dbtのfreshnessリファレンスによると、freshnessは1.9から、loaded_at_fieldは1.10からconfigの下に書く形に変わりました。ソース単位の設定はテーブル単位で上書きでき、freshness: nullで検査から外せます。periodはminute・hour・dayの3種類です。

閾値は「何時間古いと誰が困るか」から逆算します。毎朝9時の会議で前日分の売上を使うなら、3時のロード後6時間で警告、24時間で異常、という閾値の設定例です。月1回しか更新されないマスタに1日の閾値を付けると毎日鳴るため、上のproduct_masterのように外すか、日数を延ばします。

dbt source freshnessのerrorで後続ビルドを止める実行順序

実行は次のコマンドです。対象を絞るときは--selectを付けます。

dbt source freshness --select "source:ec"
cat target/sources.json

dbt sourceコマンドのページは、古いテーブルがあれば警告またはエラーを報告し、非ゼロの終了コードで終わると説明しています。前節のDAGで鮮度チェックをdbt buildより前に置いたのはこのためです。取り込みが止まった日は鮮度タスクが落ち、古いデータでの再集計が走りません。結果はtarget/sources.jsonに残るので、検査の履歴を残したい場合はこのファイルを保存します。

なお、dbtの公式ドキュメントはv2（dbt Fusion）向けに、モデルも対象にするdbt freshnessコマンドを案内しています。dbt Core 1.12系で運用している間はdbt source freshnessを使ってください。

件数の急減とスキーマ変化をSQLとdbtのcontractで検知する実装

鮮度が正常でも、届いたデータが半分しかない、列が消えている、という事故は残ります。ここからは件数とスキーマです。

前週までの同曜日平均と比べて件数の急減を検知するsingular test

前日比で判定すると、土日や月末の自然な増減で毎週鳴ります。比較の基準にするのは、直近4週の同じ曜日の平均です。tests/assert_orders_volume_not_dropped.sqlとして保存します。

{{ config(severity='error') }}

with yesterday as (
    select count(*) as n
    from {{ source('ec', 'orders') }}
    where ordered_at >= current_date - interval '1 day'
      and ordered_at < current_date
),
same_weekday as (
    select count(*) / 4.0 as avg_n
    from {{ source('ec', 'orders') }}
    where ordered_at >= current_date - interval '29 day'
      and ordered_at < current_date - interval '1 day'
      and extract(dow from ordered_at) = extract(dow from current_date - interval '1 day')
)
select yesterday.n, same_weekday.avg_n
from yesterday
cross join same_weekday
where yesterday.n < same_weekday.avg_n * 0.5

dbtのsingular testは、返した行が1行以上あれば不合格です。前日の件数が同曜日平均の半分を下回るとテストが落ち、dbt buildのタスクごと失敗するので、前節のコールバックでSlackに届きます。日別に集計してから比べる書き方だと、前日が0件のとき集計行そのものが無くなり、テストでは検知できません。上のSQLはcount(*)を直接取るため、0件の日も確実に拾えます。

倍率0.5は出発点にすぎません。2週間ほどseverity='warn'で回し、実際の振れ幅を見てからerrorに上げてください。テストの種類と使い分けはdbt testの書き方で詳しく扱っています。

information_schemaの列定義を日次で保存し差分を出すスキーマ監視SQL

取り込み側の列の増減は、DWHのメタデータで見張ります。PostgreSQLのinformation_schema.columnsは、テーブルごとの列名・順序・型を返すビューです。毎日1回、その日の列定義を保存し、前日と突き合わせます。

create table if not exists monitor.column_snapshot (
    captured_on date,
    table_name  text,
    column_name text,
    data_type   text
);

delete from monitor.column_snapshot where captured_on = current_date;

insert into monitor.column_snapshot
select current_date, table_name, column_name, data_type
from information_schema.columns
where table_schema = 'raw_ec';

select coalesce(t.table_name, y.table_name)   as table_name,
       coalesce(t.column_name, y.column_name) as column_name,
       y.data_type as before_type,
       t.data_type as after_type
from (select * from monitor.column_snapshot where captured_on = current_date) t
full outer join
     (select * from monitor.column_snapshot where captured_on = current_date - 1) y
  on t.table_name = y.table_name
 and t.column_name = y.column_name
where t.data_type is distinct from y.data_type;

最後のSELECTが返す行が、追加（before_typeがNULL）・削除（after_typeがNULL）・型変更のいずれかです。is distinct fromを使っているのは、片側がNULLの行も差分として残すためです。このビューは実行ユーザーが権限を持つ列しか返さないので、監視用ユーザーに対象スキーマの参照権限を付けておきます。

下流モデルにcontractを付けて型の不一致をビルド前に止める設定

差分SQLは「変わった」ことを知らせるだけです。下流のモデルを壊したくない場合は、dbtのcontractでビルド自体を止めます。

models:
  - name: fct_orders
    config:
      contract:
        enforced: true
    columns:
      - name: order_id
        data_type: int
      - name: ordered_at
        data_type: timestamp
      - name: amount
        data_type: numeric

dbtのcontractリファレンスによると、列名・型・列数がYAMLと一致しないと、テーブルが作られる前にコンパイルエラーで止まります。対象はtable・view・incremental（on_schema_changeがappend_new_columnsかfail）で、ephemeralやPythonモデルには付けられません。上流と合意した列定義を契約として管理する考え方は、データ契約（Data Contract）とはで掘り下げています。

アラートを鳴らしすぎないための通知設計と、専用ツールを入れる判断基準

4信号が揃うと、次に困るのは通知の多さです。鳴りすぎる監視は、いずれ誰も見なくなります。

警告と異常で通知先を分け、受け手と一次対応を名前で決める運用

通知は2段に分けます。警告（freshnessのwarn、severity=warnのテスト）は日中にまとめて見るチャンネルへ、異常（タスクの最終失敗、error）の送り先は担当者がメンションされるチャンネルです。異常には「誰が・何分以内に・何を見るか」を決め、手順を1枚にまとめます。止まった時間を測るならMTTRとはの指標がそのまま使えます。

通知文には、テーブル名・閾値・実測値・run_idの4つを入れてください。「失敗しました」だけでは、受け取った人がAirflowの画面を開くところから始めることになります。ログをJSONで残しておくと、通知から該当行への検索時間を短縮できる仕組みです。書き方は構造化ログとはにまとめています。

監視対象と閾値調整の負担から見る自前実装の規模と専用ツールへ移る条件

本記事の実装で回るのは、監視対象がおおむね数十テーブルまでで、閾値を人が見直せる規模です。次のどれかに当てはまったら、自前の延長はやめます。

閾値の見直しが週に何度も発生し、担当者の作業として回らない
値の分布（NULL率・カテゴリの偏り）まで見たいが、テストを書き切れない
異常が出たとき、影響を受けるダッシュボードを人が探している

dbt中心の基盤ならdbtパッケージ型のElementary、Python側で検証を組みたいならGreat Expectations（使い方はGreat Expectationsとは）が次の候補です。逆に、テーブルが10本未満で更新が日次1回なら、専用ツールは過剰です。鮮度と失敗通知の2つだけで止めてください。監視を載せる前に基盤の層構造から見直したい場合は、データ分析基盤構築・MLOps構築支援で現状のパイプラインの棚卸しから相談できます。

データパイプライン監視の実装でよくある質問と閾値・通知・ツール選びの判断

データパイプライン監視を組むときに出やすい質問に答えます。

dbtを使っていなくてもデータパイプライン監視は実装できますか？

できます。鮮度は「対象テーブルのmax(_loaded_at)と現在時刻の差が閾値を超えたら1行返す」SQLを、件数は本記事のsingular testからsource()を実テーブル名に置き換えたSQLを、Airflowのタスクとして定期実行すれば同じ検知ができます。結果が1行以上ならタスクを失敗させ、on_failure_callbackで通知する構成です。dbtを入れる利点は、閾値をYAMLで一元管理でき、モデルの依存関係と並べて実行できる点にあります。

鮮度の閾値は何時間に設定すればよいですか？

データの更新間隔ではなく、使う側の締め時刻から決めます。毎朝9時に前日分を使うなら、ロード完了予定から9時までの余裕が上限です。そのうえで、警告は「今気づけばまだ間に合う」時点、異常は「もう間に合わない」時点に置きます。最初の2週間は警告だけで運用し、実際のロード完了時刻の振れ幅を見てから異常の閾値を確定させると、誤報を減らせます。

Airflow 2系のsla_miss_callbackはそのまま使えますか？

Airflow 3系では使えません。SLA機能は3.0で削除され、3.1で導入されたDeadline Alertsが後継です。移行時は、sla引数で指定していた時間をDeadlineAlertのintervalに、通知処理をcallbackに置き換えます。ただしDeadline Alertsは公式に実験的機能とされているため、更新のたびに仕様を確認してください。

件数監視とスキーマ監視はどちらを先に入れるべきですか？

取り込み元が外部APIやSaaSなら件数を先に入れます。0件や半減は、API側の障害や認証切れで頻繁に起きるためです。取り込み元がCSV連携や他部署のDBで、列の追加・改名が予告なく起きる環境なら、スキーマ監視を先にします。迷う場合は、過去半年の障害記録を見て、多かったほうから着手してください。

監視の通知が多すぎて誰も見なくなった場合はどうすればよいですか？

まず1か月分の通知を、対応が必要だったものと不要だったものに仕分けます。不要だった通知の多くは、閾値が実態に合っていないか、更新頻度の低いテーブルに日次の閾値を付けているかのどちらかです。前者は閾値を実測に合わせて緩め、後者はfreshness: nullや週次の検査に移します。異常チャンネルに残すのは、受け取った人が必ず手を動かす通知だけにしてください。