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

🏗️ 架构概述

CSV数据 → Meltano (提取/加载) → DuckDB → dbt Core (转换) → 分析表

数据流:

1. 提取：Meltano的 tap-csv 读取示例员工数据
2. 加载：target-duckdb 将原始数据加载到DuckDB中
3. 转换：dbt模型创建临时视图和分析表
4. 验证：数据质量测试确保数据完整性

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

📁 项目结构

├── 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 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 .

# 仅运行分析模型
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
• 环境：开发、测试、生产

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参考 - 完整的工具文档
• 💡 使用示例 - 实用的工作流程
• 🔧 集成指南 - 详细的配置说明

🎉 状态：✅ 可用于生产环境 + MCP增强

此数据栈已通过全面测试进行了系统实现和验证。所有组件均使用最新兼容版本，并遵循最佳实践。通过集成Claude Code MCP提供智能开发协助。