Snowflake × dbt × Apache Airflowで構築するエンタープライズ本番データパイプライン完全ハンズオン
はじめに
こんにちは、NKKTech Global 技術チームです。
企業のデータ活用において、「場当たり的なSQLスクリプト」「属人化したCronジョブ」「壊れたデータの垂れ流し」は、データ組織が直面する最大の負債です。現代のエンタープライズ環境では、データパイプラインにもソフトウェア開発と同等以上のテスト可能性・バージョン管理・リネージ（系統性）・アクセス制御が求められます。
本記事では、グローバルなデータ基盤構築で標準となっている以下のスタックを用い、Rawデータの取り込みから本番マートの生成までを完全自動化・監視する実践的なデータパイプラインの構築手法をハンズオン形式で解説します。
- Snowflake: コンピュートとストレージが完全分離されたスケーラブルなクラウドDWH
- dbt (data build tool): SQLのモジュール化、データテスト、ドキュメント生成を担う変換レイヤー
- Apache Airflow (Astronomer Cosmos): 依存関係制御、再実行、障害監視を行うオーケストレーター
1. エンタープライズ・データアーキテクチャ（Medallion Architecture）
パイプラインの基本構成として、DatabricksやSnowflakeで推奨されるメダリオンアーキテクチャ（3層構造）を採用します。
[ 各種データソース (SaaS / DB / Logs) ]
│
▼ (Fivetran / Airbyte / Snowpipe)
┌──────────────────────────────────────────────────────────┐
│ Snowflake │
│ ┌────────────────────────────────────────────────────┐ │
│ │ BRONZE (RAW) : 抽出元の生データをそのまま保持 │ │
│ └────────────────────────────────────────────────────┘ │
│ │ (dbt staging models) │
│ ▼ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ SILVER (INTERMEDIATE) : クレンジング、型変換、結合 │ │
│ └────────────────────────────────────────────────────┘ │
│ │ (dbt marts models / 差分更新) │
│ ▼ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ GOLD (ANALYTICS) : BIツールやAI向け集計マート層 │ │
│ └────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────┘
▲
│ (DAGスケジュール・品質ゲート制御)
┌──────────────────────────────────────────────────────────┐
│ Apache Airflow (Astronomer Cosmos 経由でdbtモデルを実行) │
└──────────────────────────────────────────────────────────┘
2. Snowflakeの基盤設計とRBAC（権限管理）
まずはSnowflake側のウェアハウス、データベース、およびdbt実行専用のサービスロールを設定します。
-- 1. 変換専用ウェアハウスの作成 (不要時は即時サスペンドしてコスト節約)
CREATE WAREHOUSE IF NOT EXISTS TRANSFORMING_WH
WITH WAREHOUSE_SIZE = 'X-SMALL'
AUTO_SUSPEND = 60
AUTO_RESUME = TRUE
INITIALLY_SUSPENDED = TRUE;
-- 2. データベースとスキーマの作成
CREATE DATABASE IF NOT EXISTS PROD_DWH;
CREATE SCHEMA IF NOT EXISTS PROD_DWH.BRONZE;
CREATE SCHEMA IF NOT EXISTS PROD_DWH.SILVER;
CREATE SCHEMA IF NOT EXISTS PROD_DWH.GOLD;
-- 3. 最小権限の原則に基づくロール・ユーザー作成
CREATE ROLE IF NOT EXISTS DBT_PROD_ROLE;
GRANT USAGE ON WAREHOUSE TRANSFORMING_WH TO ROLE DBT_PROD_ROLE;
GRANT USAGE ON DATABASE PROD_DWH TO ROLE DBT_PROD_ROLE;
-- Bronzeは読み取りのみ、Silver/Goldは作成権限を付与
GRANT USAGE ON SCHEMA PROD_DWH.BRONZE TO ROLE DBT_PROD_ROLE;
GRANT SELECT ON ALL TABLES IN SCHEMA PROD_DWH.BRONZE TO ROLE DBT_PROD_ROLE;
GRANT SELECT ON FUTURE TABLES IN SCHEMA PROD_DWH.BRONZE TO ROLE DBT_PROD_ROLE;
GRANT ALL ON SCHEMA PROD_DWH.SILVER TO ROLE DBT_PROD_ROLE;
GRANT ALL ON SCHEMA PROD_DWH.GOLD TO ROLE DBT_PROD_ROLE;
-- サービスユーザーの作成
CREATE USER IF NOT EXISTS DBT_SERVICE_USER
PASSWORD = 'StrongServicePassword123!'
DEFAULT_ROLE = DBT_PROD_ROLE
DEFAULT_WAREHOUSE = TRANSFORMING_WH;
GRANT ROLE DBT_PROD_ROLE TO USER DBT_SERVICE_USER;
3. dbtによるデータモデリングと品質テスト
① プロジェクト設定 (dbt_project.yml)
スキーマの配置先を層ごとに明示します。
name: 'enterprise_pipeline'
version: '1.0.0'
config-version: 2
profile: 'snowflake_prod'
model-paths: ["models"]
models:
enterprise_pipeline:
staging:
+materialized: view
+schema: silver
marts:
+materialized: incremental # コスト削減のための差分更新
+schema: gold
② 差分更新モデル (models/marts/fct_orders.sql)
データ量が数億件にスケールしてもSnowflakeのクレジット消費を最小化するため、incremental（増分更新）マテリアライゼーションを使用します。
{{
config(
materialized='incremental',
unique_key='order_id',
incremental_strategy='merge'
)
}}
WITH source_orders AS (
SELECT * FROM {{ ref('stg_orders') }}
)
SELECT
order_id,
customer_id,
order_status,
total_amount,
ordered_at,
CURRENT_TIMESTAMP() AS updated_at
FROM source_orders
{% if is_incremental() %}
-- 過去実行分より新しいレコードのみをマージ処理
WHERE ordered_at > (SELECT MAX(ordered_at) FROM {{ this }})
{% endif %}
③ データ品質テスト定義 (models/marts/schema.yml)
dbtのテスト機能を用い、「主キーの重複」や「外部キー制約の破綻」を未然に防ぎます。
version: 2
models:
- name: fct_orders
description: "確定済み注文ファクトテーブル"
columns:
- name: order_id
tests:
- unique
- not_null
- name: customer_id
tests:
- not_null
- relationships:
to: ref('dim_customers')
field: customer_id
- name: total_amount
tests:
- not_null
4. Apache Airflow (Cosmos) によるオーケストレーション
従来のAirflowでは BashOperator('dbt run') が多用されていましたが、これではdbtの個別モデルがタスクとして可視化されず、リトライ制御も困難でした。
現在は、dbtプロジェクトをAirflowのDAGにネイティブパースする Astronomer Cosmos がデファクトスタンダードです。
必要なライブラリ
pip install apache-airflow-providers-snowflake astronomer-cosmos
DAG定義 (dags/dbt_snowflake_dag.py)
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping
# Snowflake接続設定 (AirflowのConnection: snowflake_default を利用)
profile_config = ProfileConfig(
profile_name="snowflake_prod",
target_name="prod",
profile_mapping=SnowflakeUserPasswordProfileMapping(
conn_id="snowflake_default",
profile_args={
"database": "PROD_DWH",
"schema": "GOLD"
}
)
)
default_args = {
"owner": "data_engineering",
"retries": 2,
"retry_delay": timedelta(minutes=3),
"email_on_failure": True,
}
# Cosmosにより、dbtの全モデルとテストがAirflowの個別タスクとしてDAG展開される
dbt_pipeline_dag = DbtDag(
project_config=ProjectConfig("/opt/airflow/dbt"),
profile_config=profile_config,
execution_config=ExecutionConfig(dbt_executable_path="/usr/local/bin/dbt"),
operator_args={
"install_deps": True,
"full_refresh": False,
},
dag_id="snowflake_dbt_enterprise_pipeline",
schedule_interval="@daily",
start_date=datetime(2026, 1, 1),
catchup=False,
default_args=default_args,
tags=["snowflake", "dbt", "gold_marts"]
)
5. 本番運用のベストプラクティス
エンタープライズ本番環境で長期的にシステムを安定稼働させるための重要ポイントです。
① Zero-Copy Cloneを活用したCI/CD
プルリクエスト（PR）作成時、GitHub ActionsでSnowflakeのZero-Copy Cloneを実行します。
-- 本番データを複製することなく、一瞬で同一ストレージの検証用DBを作成
CREATE DATABASE DEV_PR_123 CLONE PROD_DWH;
この一時DBに対して dbt build を実行することで、ストレージコストゼロで本番同等の完全な結合テストを行えます。テスト完了後はDBを破棄（DROP DATABASE）します。
② Snowflakeコストの暴走防止（リソースモニター）
万が一Airflowのループや重いクエリが回った場合に備え、クレジット上限でウェアハウスを強制停止するリソースモニターを必ず設定します。
CREATE RESOURCE MONITOR IF NOT EXISTS LIMIT_DAILY_CREDITS
WITH CREDIT_QUOTA = 50
FREQUENCY = DAILY
TRIGGERS ON 80% DO NOTIFY
ON 100% DO SUSPEND_IMMEDIATE;
ALTER WAREHOUSE TRANSFORMING_WH SET RESOURCE_MONITOR = LIMIT_DAILY_CREDITS;
まとめ：モダンデータスタックで手に入れる「信頼性」
Snowflakeのコンピュート柔軟性、dbtのソフトウェア工学的アプローチ、そしてAirflowによる堅牢なスケジューリングが揃うことで、以下が実現します。
- 品質の担保: テストが失敗したデータはGold層へ到達させない（Bad Dataの防御）。
- コストの最小化: 差分更新とウェアハウスの自動停止によるクレジット最適化。
- 運用属人化の解消: dbtリネージグラフとAirflow DAG画面による圧倒的な可観測性。
NKKTech Globalでは、データ基盤のアーキテクチャ設計からクラウドDWHのパフォーマンスチューニング、ML基盤との統合までをグローバル体制でご支援しています。
お問い合わせ先
Snowflake導入、dbtへの移行、データパイプラインの刷新、コスト最適化に関するご相談は、下記よりお気軽にお問い合わせください。
- Webサイト: https://nkktech.com/
- メール: contact@nkk.com.vn
- LinkedIn: https://www.linkedin.com/company/nkktech
著者：NKKTech Global 技術チーム
私たちは、最先端のデータエンジニアリングとクラウドアーキテクチャで、ビジネスの意思決定を支える強固なデータ基盤を構築します。