AI

データパイプライン自動化の実装：Airflowのアセット起動・イベント駆動とdbtのCIで手作業をなくす手順【2026年版】

データパイプライン自動化の実装：Airflowのアセット起動・イベント駆動とdbtのCIで手作業をなくす手順【2026年版】

データパイプライン自動化で人の手から外すべきものは、ジョブの実行ボタンだけではありません。「上流が終わったか確認してから流す」「落ちたら朝に再実行する」「SQLを直したら本番で試す」という3つの作業が残っている限り、自動化はまだ半分までの状態です。本記事では、Airflow 3.4系とdbt Core 1.12系（いずれも2026年11月時点の最新）を使い、時刻起動・データ起動・イベント起動・変更起動の4つをコード付きで実装します。最後に、オーケストレーターを入れずにcronで済ませるべき条件も示します。

まとめ：データパイプライン自動化は時刻起動から始めデータ起動とCIへ広げる

最初に入れるのは、リトライ付きの時刻起動です。Airflowのretriesとretry_exponential_backoffを設定するだけで、一時的な接続エラーによる朝の手動再実行はほぼ消えます。

次に、cronの時刻ずらしで上流と下流をつないでいる箇所をAssetによるデータ起動へ置き換えます。ファイル到着が不定期な取り込みにはS3からSQS経由のイベント起動を、SQLの変更にはdbtのstate:modified+で変更分だけを検証するCIを足してください。テーブルが数本で日次1回の更新なら、ここまでは不要です。cronと1本のスクリプトで止めます。

データパイプライン自動化で人の手から外す4つの起点とcronだけで組んだ構成の事故

実装の前に、どの作業を機械に渡すのかを棚卸しします。抜けやすいのは、実行そのものより「実行してよいかの判断」です。

時刻・データ・イベント・変更の4起点で手作業の残る箇所を棚卸しする表

パイプラインの起動のきっかけは4種類に分けられます。本記事で実装する範囲と対応させると次のとおりです。変更起動の行では、dbtのstate:modified+による変更分の選択とGitHub ActionsでのCI実行を組み合わせます。

起点 残りがちな手作業 自動化の手段 本記事の実装
時刻 失敗時の手動再実行 スケジュール＋リトライ retries・backoff
データ 上流の完了確認 上流の出力を合図に起動 Asset（outlets）
イベント ファイル到着の確認 外部キューのメッセージで起動 S3通知＋SQS＋AssetWatcher
変更 SQL修正後の本番確認 プルリクエストで検証 dbt＋GitHub Actions

取込・変換・配信という層の分け方そのものはデータパイプラインとはで整理しています。本記事は、その各層を「誰も触らずに動く」状態にする手順に絞ります。

cronの時刻ずらしで依存をつなぐ構成が上流の遅延で静かに壊れる仕組み

よくある構成は、取り込みを2時、変換を3時、集計を4時にcronで並べる形です。上流が1時間遅れた日、変換は前日のデータのまま3時に成功します。終了コードは0なので、誰も気づきません。

時刻の間隔を広げても解決しません。遅延の上限は誰にも分からず、間隔を広げた分だけ毎日の鮮度が落ちます。必要なのは「上流が書き終えたこと」を起動条件にする仕組みで、それが次章以降のデータ起動とイベント起動です。壊れたことを検知する側の実装はデータパイプライン監視の実装で扱っています。

検証環境を整えAirflow 3.4系のスケジュールとリトライで日次処理を自動化する手順

最初の自動化は時刻起動です。DAGの基本的な書き方と環境構築はApache Airflowとはで済ませている前提で進めます。

Airflow 3.4.0とdbt Core 1.12.6を入れる検証環境の構築コマンド

PyPIのapache-airflowでは2026年11月6日公開の3.4.0が、dbt-coreでは10月14日公開の1.12.6が最新です。イベント起動で使うメッセージキュー用のproviderも、この時点で一緒に入れます。

python -m venv .venv
source .venv/bin/activate
pip install "apache-airflow==3.4.0" \
  "apache-airflow-providers-common-messaging[amazon]" \
  "dbt-core==1.12.6" dbt-postgres
airflow standalone

common.messagingの公式ページでは、このproviderが対応する最小のAirflowは3.0.1で、amazon extraを付けるとAmazon providerが一緒に入ると案内されています。本番ではAirflow公式の制約ファイルを使い、依存の版を固定してください。

retriesとcatchupを明示した日次DAGで朝の手動再実行をなくすコード

次のDAGは毎日2時（日本時間）に取り込みとdbtのビルドを流します。一時的なエラーは間隔を延ばしながら3回まで再試行し、過去分の一斉実行は止めています。

from datetime import timedelta

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

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
}

with DAG(
    dag_id="daily_orders",
    schedule="0 2 * * *",
    start_date=pendulum.datetime(2026, 11, 1, tz="Asia/Tokyo"),
    catchup=False,
    max_active_runs=1,
    default_args=default_args,
):
    extract = BashOperator(
        task_id="extract_orders",
        bash_command="python /opt/jobs/extract_orders.py --date {{ ds }}",
    )
    transform = BashOperator(
        task_id="dbt_build",
        bash_command="cd /opt/dbt && dbt build --select tag:orders",
    )
    extract >> transform

押さえる設定は3つです。catchup=Falseは、停止していた期間の実行を再開時にまとめて流さないための指定で、既定値に頼らない明示的な指定です。max_active_runs=1は前日の実行が終わる前に翌日分が走るのを防ぎます。取り込みスクリプトに{{ ds }}（論理日付）を渡しておくと、特定日だけの再実行がUIから1クリックで済みます。

Assetで上流の更新完了を合図に下流を動かすデータ起動の自動化を実装する

cronの時刻ずらしを、上流の書き込み完了を合図にする起動へ置き換えます。Airflow 3系ではこの仕組みをAssetと呼びます。

outletsとschedule=[asset]で取込・変換DAGをつなぐ実装コード

Asset-Aware Schedulingの公式ドキュメントの形に沿って、取り込み側のタスクにoutletsを、変換側のDAGのscheduleに同じAssetを指定します。

import pendulum
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG, Asset

raw_orders = Asset("s3://raw-bucket/orders/")

with DAG(
    dag_id="ingest_orders",
    schedule="0 2 * * *",
    start_date=pendulum.datetime(2026, 11, 1, tz="Asia/Tokyo"),
    catchup=False,
):
    BashOperator(
        task_id="load_raw_orders",
        bash_command="python /opt/jobs/load_orders.py",
        outlets=[raw_orders],
    )

with DAG(
    dag_id="transform_orders",
    schedule=[raw_orders],
    start_date=pendulum.datetime(2026, 11, 1, tz="Asia/Tokyo"),
    catchup=False,
):
    BashOperator(
        task_id="dbt_build_orders",
        bash_command="cd /opt/dbt && dbt build --select source:raw.orders+",
    )

load_raw_ordersが成功した時点でAssetの更新が記録され、transform_ordersが起動します。取り込みが失敗した日は変換も走らないため、前日データのまま成功する事故が構造的に起きません。Assetの文字列はURIの形をとりますが、Airflowはその場所を見に行くわけではなく、更新の記録を照合する識別子として扱います。

複数の上流がそろったら起動するAND条件とどちらかで起動するOR条件の書き分け

集計テーブルが受注と在庫の両方に依存するなら、2つのAssetがそろうまで待たせます。公式ドキュメントの記法では&がAND、|がORです。

orders = Asset("s3://raw-bucket/orders/")
stock = Asset("s3://raw-bucket/stock/")
returns = Asset("s3://raw-bucket/returns/")

# 受注と在庫の両方が更新されたら起動
with DAG(dag_id="build_sales_mart", schedule=(orders & stock), ...):
    ...

# 受注か返品のどちらかが更新されたら起動
with DAG(dag_id="refresh_order_events", schedule=(orders | returns), ...):
    ...

使い分けの基準は、下流が「全部そろわないと誤った数字を出すか」です。売上と在庫を突き合わせる集計はAND、イベントを追記するだけのテーブルはORで足ります。上流が5本を超えてAND条件が長くなったら、依存の設計そのものを見直す時期です。アセットを中心に依存を定義する考え方を突き詰めたツールとしてDagsterがあり、選び方は最後の章で触れます。

S3へのファイル到着をSQS経由で受けてDAGを起動するイベント駆動の実装

取引先がCSVを置く時刻が日によって違うなら、時刻起動もデータ起動も合いません。ファイル到着そのものを起点にします。

S3イベント通知をSQSへ送りAssetWatcherで待ち受ける設定とDAGのコード

手順は3段です。SQSのアクセスポリシーでS3からの送信を許可し、バケットに通知設定を入れ、Airflow側でキューを監視します。権限のポリシー例はS3の公式ドキュメントにあり、Principalにs3.amazonaws.com、条件にaws:SourceArnとaws:SourceAccountを指定する形です。通知設定は次のとおりです。

cat > notify.json <<'EOF'
{
  "QueueConfigurations": [
    {
      "QueueArn": "arn:aws:sqs:ap-northeast-1:123456789012:orders-arrival",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": "incoming/"}]}}
    }
  ]
}
EOF
aws s3api put-bucket-notification-configuration \
  --bucket partner-upload-bucket \
  --notification-configuration file://notify.json

S3の通知設定の解説によると、設定の反映にはおよそ5分かかり、有効化した直後にs3:TestEventが1通届きます。Airflow側はcommon.messagingの公式例と同じく、MessageQueueTriggerをAssetの監視役に付けます。

from airflow.providers.common.messaging.triggers.msg_queue import MessageQueueTrigger
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG, Asset, AssetWatcher

trigger = MessageQueueTrigger(
    scheme="sqs",
    sqs_queue="https://sqs.ap-northeast-1.amazonaws.com/123456789012/orders-arrival",
)
partner_files = Asset(
    "partner_orders_arrival",
    watchers=[AssetWatcher(name="orders_arrival_watcher", trigger=trigger)],
)

with DAG(dag_id="load_partner_orders", schedule=[partner_files]):
    BashOperator(
        task_id="load_new_files",
        bash_command="python /opt/jobs/load_partner_files.py --prefix incoming/",
    )

AWSへの接続には既定の接続IDaws_defaultが使われます。このイベント駆動スケジューリングはAirflowの公式ドキュメントで3.0から追加された機能と明記されており、2系の環境では使えません。最初に届くs3:TestEventでもDAGが1回起動するので、取り込み処理は「対象ファイルが無ければ何もせず正常終了する」作りにしておきます。

イベント駆動で起きる二重起動と再配信に備える冪等な書き込みの型

S3のイベント通知は「少なくとも1回」届く設計で、通常は数秒、ときに1分以上遅れると公式に書かれています。同じファイルの通知が2回届くことを前提とした実装が必要です。Airflow側でも、トリガーの再起動後にブローカーが再配信する場合があるため、受け手を冪等にするよう公式ドキュメントが求めています。

実装の型は2つです。1つは、ファイルのキーとETagを取り込み済み台帳のテーブルに記録し、記録済みなら読み飛ばす方法。もう1つは、取り込み先を日付パーティションごとに上書きする方法で、何度流しても結果が同じになります。追記型のテーブルにそのままINSERTする設計は、イベント駆動とは組み合わせないでください。なお、S3の通知先にFIFOキューは指定できません。順序を保証したい場合はEventBridge経由にするとAWSは案内しています。

dbtのstate:modified+とGitHub Actionsで変更分を検証するCI

4つ目の起点は、SQLの変更です。変換処理を直した人が本番で流して確かめる運用は、それ自体が手作業の自動化漏れです。dbtのプロジェクト構成とCIの基本はdbt Coreの使い方にまとめています。

本番のmanifestと比べて変更モデルと下流だけをビルドするコマンド

dbtは、前回の実行で出力した成果物（target/配下のmanifest）と今のコードを比べ、新規・変更されたモデルを選べます。dbtの公式ドキュメントの形に沿うと、コマンドは次のとおりです。

# 本番ジョブの最後に成果物を保存しておく
aws s3 cp target/manifest.json s3://dbt-artifacts/prod/manifest.json

# CIでは本番のmanifestを取得し、変更モデルとその下流だけをビルドする
aws s3 cp s3://dbt-artifacts/prod/manifest.json prod-artifacts/manifest.json
dbt build --select state:modified+ --defer --state prod-artifacts --target ci

--deferを付けると、変更していない上流モデルはCI用スキーマに作らず本番のテーブルを参照します。deferの公式解説では、この組み合わせの呼び名は「Slim CI」です。モデルが200本あってもCIで作るのは変更した数本とその下流だけになり、実行時間とDWHの費用が下がります。--stateに渡す成果物は、実行中のdbtと互換性のあるスキーマの版が条件です。本番とCIでdbtの版をそろえてください。

プルリクエストで変換処理を検証するGitHub Actionsのワークフロー定義

上のコマンドを、プルリクエストのたびに自動で走らせます。接続情報はリポジトリのSecretsに入れ、profiles.ymlから環境変数で読ませる前提です。

name: dbt-slim-ci
on:
  pull_request:
    paths: ["models/**", "macros/**", "dbt_project.yml"]

jobs:
  build-changed:
    runs-on: ubuntu-latest
    env:
      DBT_CI_PASSWORD: ${{ secrets.DBT_CI_PASSWORD }}
      AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
      AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install "dbt-core==1.12.6" dbt-postgres
      - run: aws s3 cp s3://dbt-artifacts/prod/manifest.json prod-artifacts/manifest.json
      - run: dbt deps
      - run: dbt build --select state:modified+ --defer --state prod-artifacts --target ci

テストが1本でも落ちればプルリクエストに赤い印が付き、マージ前に止まります。どのテストを書くかはdbt testの書き方で、generic・singular・unit testの使い分けとして整理しています。CI用スキーマはプルリクエストごとに名前を分けると、同時に複数の変更を検証しても衝突しません。

GitHub Actionsのscheduleで本体を起動しない理由と起動処理の配置先

CIと同じ場所で本番の定時実行も済ませたくなりますが、勧めません。GitHubの公式ドキュメントは、scheduleイベントについて次の制約を明記しています。

  • 高負荷時、特に毎時0分前後は実行が遅れ、負荷が十分に高いとキューのジョブが破棄されることがある
  • 最短の間隔は5分で、時刻はUTC基準
  • 既定ブランチでしか動かない
  • 公開リポジトリでは60日間活動が無いと自動で無効になる

「破棄されることがある」起動は、業務データの定時更新には使えません。GitHub Actionsは変更起動（プルリクエストのCI）に限定し、時刻・データ・イベント起動はAirflowなどのオーケストレーターに置く、という役割分担で固定します。

自動化の範囲を決める判断基準とオーケストレーターを入れずに済ませる条件

4つの起点をすべて自動化する必要はありません。データの届き方と規模で、入れる範囲を決めます。

時刻起動・データ起動・イベント起動のどれを選ぶかを決める判断表

起動方式は、上流の終わり方とデータの届き方で決まります。

状況 選ぶ起動方式 理由
外部APIから決まった時刻に取得 時刻起動＋リトライ 取得のタイミングを自社で決められる
社内の上流ジョブの完了後に変換 データ起動（Asset） 上流の遅延にそのまま追従できる
取引先のファイル到着が不定期 イベント起動（SQS） 到着時刻を予測できない
秒単位の鮮度が必要 ストリーミング基盤 DAGの起動単位では間に合わない

最後の行に当たる場合、Airflowのイベント起動は解になりません。DAGの起動と終了の間隔は秒単位の要件に向かないため、Kafkaなどのストリーミング処理を別に設計します。処理方式の選び分けはデータパイプラインとはのバッチ・ストリーミング・CDCの章を参照してください。

テーブルが数本で日次1回ならcronと1本のスクリプトで止めるべき場面

Airflowを本番で動かすと、スケジューラ・Webサーバ・メタデータDBの3つを保守する仕事が増えます。次の条件が全部そろうなら、オーケストレーターは過剰です。

  • 連携するテーブルが10本未満で、上流と下流の依存が1段だけ
  • 更新は日次1回で、遅れても当日中に直せば業務が回る
  • データの担当者が1人で、変更の頻度が月に数回

この規模では、cronから1本のシェルスクリプトを呼び、取り込みとdbtのビルドを&&でつなげば足ります。上流が失敗したら後続が走らない点は、Assetと同じ効果です。失敗時に通知する処理だけは必ず入れてください。依存が2段を超えた、ファイル到着が不定期になった、担当者が2人を超えた、のどれかが起きた時点がオーケストレーターへ移る目安です。cronから呼ぶ取込・変換・配信のスクリプトの組み方はデータパイプライン構築の手順で解説しています。

Dagster・Step Functionsを選ぶ条件と外部に任せる範囲の線引き

オーケストレーターを入れるとして、Airflowが常に正解ではありません。テーブル単位の依存と品質チェックを中心に管理したいならDagsterが、処理がAWSのLambdaやECSタスクで完結するならAWS Step Functionsが、サーバの保守を減らせます。Pythonの関数をそのままフローにしたい小規模チームにはPrefectも候補です。Airflowを選ぶのは、既存のDAG資産がある場合と、接続先が多くproviderの数が効く場合です。

外部に任せる範囲は、オーケストレーターの基盤構築と、本記事のような起動方式の設計までにします。どのテーブルをどの鮮度で更新するかという業務側の判断は、社内に残してください。既存のcronやスクリプトの棚卸しから自動化の範囲を決めたい場合は、データ分析基盤構築・MLOps構築支援で相談を受け付けています。

データパイプライン自動化の実装でよくある質問と起動方式・ツール選びの判断

データパイプラインの自動化を進めるときに出やすい質問に答えます。

データパイプラインの自動化とオーケストレーションは何が違いますか？

自動化は「人が手でやっていた作業を機械に任せること」全般を指し、オーケストレーションはそのうち複数の処理の順序・依存・再実行を管理する部分です。cronで1本のスクリプトを定時実行するのは自動化ですが、オーケストレーションとは呼びません。本記事のAsset起動やリトライはオーケストレーションの機能、dbtのCIはその外側の自動化です。概念の違いはオーケストレーションとはで詳しく扱っています。

Airflow 2系の環境でもAssetやイベント起動は使えますか？

Airflow 2系では、同じ仕組みが「Dataset」という名前で2.4から提供されており、outletsとscheduleによるデータ起動は書けます。3系でAssetへ改名され、importもairflow.sdkに移りました。一方、AssetWatcherとMessageQueueTriggerによるイベント起動は3.0で追加された機能で、2系では使えません。S3のファイル到着で起動したい場合は、3系へ移行するか、LambdaからAirflowのREST APIを呼んでDAGを起動する構成で代替します。

ファイル到着の検知はセンサーとイベント起動のどちらを使うべきですか？

到着時刻がおおむね決まっていて、待つのが数十分程度ならセンサーでも足ります。到着が不定期で、1日中待つ可能性があるならイベント起動を選んでください。センサーは待っている間もタスクとして枠を使い続け、待ち時間の上限を決める必要があります。イベント起動はメッセージが届くまでDAG自体が起動しないため、待ち時間の設計が要りません。ただし本記事のとおり、二重に届く前提の冪等な取り込みが条件です。

dbtのCIで本番のmanifestが無い初回はどうすればよいですか？

初回は--stateを付けずにdbt buildで全モデルをCI用スキーマに作り、同時に本番ジョブの最後でmanifestを保存する設定を入れます。2回目以降は保存したmanifestと比較できるようになります。manifestの保存先は、本番ジョブだけが書き込めて、CIは読み取りだけできる権限に分けてください。CIが本番の比較元を書き換えると、変更の検出が狂います。

自動化したあとに失敗に気づけない状態をどう防げばよいですか？

自動化で人が見なくなるぶん、異常を知らせる仕組みが要ります。最低限、タスクの最終失敗をチャットへ通知し、取り込みテーブルの鮮度を検査してください。リトライで回復した失敗も日次でまとめて確認すると、接続先の劣化に早く気づけます。Airflowの失敗通知とdbtの鮮度・件数・スキーマの検知をコード付きでまとめた手順は、データパイプライン監視の実装を参照してください。

関連記事

お気に入りに入れた記事の一覧

一創の公式 SNS

X では毎朝の新着記事と IT ニュースの速報を、YouTube・TikTok・Instagram・LinkedIn・Facebook では記事の解説動画を配信しています。

この記事は以下の記事からリンクされています

この記事に関連するサービス

資料請求

RELATED POSTS 関連記事

一覧へ戻る

目次