Claude Data Stack MCP

🚀 Meltano、DuckDB、dbt Coreを用いたモダンなデータスタック

このプロダクトは、ELTオーケストレーションにMeltano、高性能分析にDuckDB、データ変換にdbt Coreを組み合わせた、本番環境で使用可能なデータスタックの実装です。

📐 アーキテクチャの概要

CSVデータ → Meltano (抽出/ロード) → DuckDB → dbt Core (変換) → 分析テーブル

データフロー:

1. 抽出: Meltanoの tap-csv がサンプルの従業員データを読み取ります。
2. ロード: target-duckdb が生データをDuckDBにロードします。
3. 変換: dbtモデルがステージングビューと分析テーブルを作成します。
4. 検証: データ品質テストによりデータの整合性が確保されます。

🛠️ ツールのバージョン (最新の互換性のあるバージョン)

• Meltano 3.8.0 - ELTパイプラインのDataOpsプラットフォーム
• DuckDB 1.3.2 - 高性能なインプロセス分析データベース
• dbt Core 1.10.4 - データ変換フレームワーク
• dbt-duckdb 1.9.4 - dbt用のDuckDBアダプター

🚀 クイックスタート

1. 環境のセットアップ

# プロジェクトをクローンして移動
cd claude-data-stack-mcp

# 仮想環境を作成してアクティブ化
python -m venv venv
source venv/bin/activate  # Windowsの場合: venv\Scripts\activate

# すべての依存関係をインストール
pip install -r requirements.txt

2. 完全なパイプラインを実行

# ステップ1: Meltanoで抽出とロード
meltano run tap-csv target-duckdb

# ステップ2: dbtで変換
cd transform
DBT_PROFILES_DIR=./profiles/duckdb dbt run --profile data_stack --project-dir .

# ステップ3: データ品質テストで検証
DBT_PROFILES_DIR=./profiles/duckdb dbt test --profile data_stack --project-dir .

3. 結果を確認

# 変換されたデータを確認
python -c "
import duckdb
conn = duckdb.connect('data/warehouse/data_stack.duckdb')
print('=== 部署統計 ===')
result = conn.execute('SELECT * FROM main.agg_department_stats').fetchall()
for row in result: print(row)
"

📁 プロジェクト構造

├── data/
│   ├── sample_data.csv          # サンプルの従業員データ
│   └── warehouse/               # DuckDBデータベースファイル
├── transform/                   # dbtプロジェクト
│   ├── models/
│   │   ├── staging/             # ステージングモデル
│   │   │   ├── stg_employees.sql
│   │   │   └── sources.yml
│   │   └── marts/               # 分析モデル
│   │       ├── dim_employees.sql
│   │       └── agg_department_stats.sql
│   ├── profiles/duckdb/         # プロジェクト内のプロファイル
│   └── dbt_project.yml
├── meltano.yml                  # Meltanoの設定
├── requirements.txt             # Pythonの依存関係
└── README.md                    # このファイル

📊 データモデル

ステージングレイヤー

• stg_employees: 生のCSVからクリーンで型付けされた従業員データ

分析レイヤー

• dim_employees: 給与階層 (ジュニア/中級/上級) を持つ従業員ディメンション
• agg_department_stats: 部署レベルの集計 (件数、平均/最小/最大給与、総給与)

データ品質テスト

• 一意性制約: 従業員IDは一意でなければなりません。
• 非null制約: 従業員IDはnullにすることはできません。

💻 使用例

新しいデータソースを追加

# 利用可能な抽出器を閲覧
meltano discover extractors

# 新しい抽出器を追加 (例: PostgreSQL)
meltano add extractor tap-postgres

# meltano.ymlで設定して実行
meltano run tap-postgres target-duckdb

新しいdbtモデルを作成

-- transform/models/marts/new_model.sql
{{ config(materialized='table') }}

select
    department,
    count(*) as employee_count,
    avg(annual_salary) as avg_salary
from {{ ref('stg_employees') }}
group by department

開発ワークフロー

# 個々のdbtモデルをテスト
cd transform
DBT_PROFILES_DIR=./profiles/duckdb dbt run --models stg_employees --profile data_stack --project-dir .

# martsモデルのみを実行
DBT_PROFILES_DIR=./profiles/duckdb dbt run --models marts --profile data_stack --project-dir .

# ドキュメントを生成
DBT_PROFILES_DIR=./profiles/duckdb dbt docs generate --profile data_stack --project-dir .

⚙️ 設定

Meltanoの設定

• 抽出器: tap-csv は data/sample_data.csv 用に設定されています。
• ローダー: target-duckdb は data/warehouse/data_stack.duckdb 用に設定されています。
• 環境: dev、staging、prod

dbtの設定

• プロファイル: プロジェクト内のプロファイルを持つ data_stack
• ターゲット: data/warehouse/ 内のDuckDBデータベース
• マテリアライゼーション: ステージングはビュー、martsはテーブル

🛠️ トラブルシューティング

一般的な問題

「テーブルが存在しません」エラー:

• MeltanoのELTステップが正常に完了したことを確認してください。
• data/warehouse/data_stack.duckdb が存在することを確認してください。

dbtプロファイルエラー:

• transform/ ディレクトリにいることを確認してください。
• DBT_PROFILES_DIR=./profiles/duckdb フラグを使用してください。

Pythonの依存関係の競合:

• 新しい仮想環境を使用してください。
• Python 3.13以上の互換性を確認してください。

検証コマンド

# Meltanoの設定を確認
meltano config list

# dbtのセットアップを検証
cd transform
DBT_PROFILES_DIR=./profiles/duckdb dbt debug --profile data_stack

# DuckDBを直接調査
python -c "import duckdb; conn = duckdb.connect('data/warehouse/data_stack.duckdb'); print(conn.execute('SHOW TABLES').fetchall())"

🚀 次のステップ

1. さらに多くのデータソースを追加: Meltanoの豊富な抽出器ライブラリを使用して、API、データベース、またはファイルを統合します。
2. 変換を拡張: 高度な分析用のより洗練されたdbtモデルを作成します。
3. オーケストレーションを追加: Airflow、Prefect、または他のオーケストレーションツールと統合します。
4. 監視を有効化: データ品質の監視とアラートを追加します。
5. ストレージをスケールアップ: クラウドデータウェアハウス (Snowflake、BigQueryなど) に移行します。

🤝 MCP統合

新機能: Claude Code MCPサーバーによるインテリジェントなdbt支援！

# Claude Code統合用のdbt MCPサーバーを起動
./scripts/start_dbt_mcp.sh

機能:

• dbt CLI操作: dbt_run、dbt_test、dbt_compile、dbt_build
• プロジェクトの探索: モデルの一覧表示、メタデータ分析、リネージ探索
• データベースクエリ: DuckDBウェアハウスに対する直接的なSQL実行
• リアルタイム支援: コンテキストを考慮したdbtプロジェクトのサポート

ドキュメント:

• 🚀 クイックスタートガイド - 5分でのセットアップ
• 📚 APIリファレンス - 完全なツールドキュメント
• 💡 使用例 - 実用的なワークフロー
• 🔧 統合ガイド - 詳細な設定

📈 ステータス: ✅ 本番環境での使用可能 + MCP強化

このデータスタックは、体系的に実装され、包括的なテストを通じて検証されています。すべてのコンポーネントは最新の互換性のあるバージョンを使用しており、ベストプラクティスに従っています。Claude Code MCPの統合により、インテリジェントな開発支援が強化されています。