Claude Data Stack MCP

🚀 現代數據棧：結合Meltano、DuckDB和dbt Core

本項目是一個可用於生產環境的數據棧實現方案，它結合了用於ELT編排的Meltano、具備高性能分析能力的DuckDB，以及用於數據轉換的dbt Core。

🚀 快速開始

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)
"

✨ 主要特性

• 集成了Meltano、DuckDB和dbt Core，構建了一個完整的生產級數據棧。
• 提供了清晰的數據處理流程，包括提取、加載、轉換和驗證。
• 支持添加新的數據來源和創建新的dbt模型，具有良好的擴展性。
• 增強了Claude Code MCP集成，提供智能開發輔助。

📦 安裝指南

工具版本（最新兼容版本）

💻 使用示例

基礎用法

添加新數據源

# 瀏覽可用的提取器
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 .

📚 詳細文檔

項目結構

├── 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必須唯一。
• 非空約束：員工ID不能為空。

配置

Meltano配置

• 提取器：tap-csv 配置為讀取 data/sample_data.csv。
• 加載器：target-duckdb 配置為寫入 data/warehouse/data_stack.duckdb。
• 環境：開發、測試、生產。

dbt配置

• 配置文件：data_stack，包含項目內的配置。
• 目標：data/warehouse/ 中的DuckDB數據庫。
• 物化方式：暫存層使用視圖，分析層使用表。

故障排除

常見問題

“表不存在”錯誤：

• 確保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參考 - 完整的工具文檔
• 💡 使用示例 - 實際工作流程
• 🔧 集成指南 - 詳細的配置說明

🔧 技術細節

架構概述

CSV數據 → Meltano (提取/加載) → DuckDB → dbt Core (轉換) → 分析表

數據流：

1. 提取：Meltano的 tap-csv 讀取示例員工數據。
2. 加載：target-duckdb 將原始數據加載到DuckDB中。
3. 轉換：dbt模型創建暫存視圖和分析表。
4. 驗證：數據質量測試確保數據完整性。

📄 許可證

文檔中未提及許可證相關信息。

本數據棧已經過系統實現和全面測試驗證。所有組件均使用最新兼容版本，並遵循最佳實踐。通過Claude Code MCP集成，提供智能開發輔助。