EMR MCP服务器
一个全面的模型上下文协议（MCP）服务器，为EMR集群管理、配置建议和监控功能提供智能指导。该服务器在EMR主节点上运行，提供对集群性能、成本优化和配置调优的实时洞察。
🚀 特性
🏗️ 群集管理
- 实时集群信息 具有详细的实例组分析
- 多集群支持 具有过滤和搜索功能
- 成本分析和估算 按实例类型细分
- 实例类型建议 基于工作负载模式
- 自动缩放策略建议 优化资源利用
📊 资源监控
- YARN资源管理器集成 用于应用程序监控
- HDFS名称节点监控 用于存储健康和利用率
- 实时资源利用率 跨所有群集节点
- 应用程序性能分析 具有瓶颈识别功能
- 历史趋势分析 用于容量规划
🧠 分析和优化
- Spark历史服务器集成 详细的工作分析
- 配置建议 基于工作负载模式
- 性能诊断 具有可操作的见解
- 成本优化建议 包括spot实例使用情况
- 特定于工作负载的调优 适用于批处理、流式处理和机器学习工作负载
🔒 安全和身份验证
- 多种身份验证方法：API密钥、JWT令牌、IAM角色
- 基于角色的访问控制 具有细粒度权限
- 安全通信 使用HTTPS和证书验证
- 请求速率限制 防止滥用
- 审核日志记录 用于合规和监控
📋 快速开始
先决条件
- EMR集群运行版本6.0+
- Python 3.8+
- 访问YARN资源管理器（端口8088）
- 访问Spark历史服务器（端口18080）
- 访问HDFS名称节点（端口9870）
安装
# Clone the repository
git clone https://github.com/your-org/emr-mcp-server.git
cd emr-mcp-server
# Install dependencies
pip install -r requirements.txt
# Configure the server
cp config/server_config.yaml.example config/server_config.yaml
# Edit the configuration file with your EMR cluster details配置
编辑 config/server_config.yaml:
server:
host: "0.0.0.0"
port: 3000
debug: false
workers: 4
emr:
region: "us-east-1"
cluster_id: "j-XXXXXXXXX" # Optional: specific cluster ID
yarn:
resource_manager_url: "http://localhost:8088"
timeout: 30
spark:
history_server_url: "http://localhost:18080"
timeout: 30
hdfs:
namenode_url: "http://localhost:9870"
timeout: 30
auth:
method: "api_key" # Options: api_key, jwt, iam
api_keys:
- "emr-mcp-default-key"
jwt_secret: "your-jwt-secret"
logging:
level: "INFO"
format: "console" # Options: console, json运行服务器
# Start the server directly
python -m src.server
# Or use the startup script
./scripts/start_server.sh
# Check server status
curl http://localhost:3000/health🛠️ MCP工具
群集管理工具
get_cluster_info
检索全面的EMR集群信息，包括配置、实例组和成本分析。
{
"name": "get_cluster_info",
"arguments": {
"cluster_id": "j-XXXXXXXXX" // Optional
}
}list_clusters
列出所有具有可选状态过滤的EMR集群。
{
"name": "list_clusters",
"arguments": {
"states": ["RUNNING", "WAITING"] // Optional
}
}estimate_cost
计算当前和预计成本，并进行详细细分。
{
"name": "estimate_cost",
"arguments": {
"runtime_hours": 48.0, // Optional
"cluster_id": "j-XXXXXXXXX" // Optional
}
}suggest_instance_types
根据工作负载特征获取基于AI的实例类型建议。
{
"name": "suggest_instance_types",
"arguments": {
"workload_type": "memory_intensive", // Options: general, compute_intensive, memory_intensive, storage_intensive
"data_size_gb": 1000, // Optional
"concurrent_jobs": 10 // Optional
}
}监视工具
monitor_resources
跨YARN、HDFS和集群节点获取实时资源利用率。
{
"name": "monitor_resources",
"arguments": {}
}analyze_yarn_applications
使用性能指标和资源使用情况分析YARN应用程序。
{
"name": "analyze_yarn_applications",
"arguments": {
"states": ["RUNNING", "FINISHED"], // Optional
"application_types": ["SPARK"], // Optional
"limit": 50 // Optional, default: 50
}
}diagnose_performance
识别性能瓶颈并获得优化建议。
{
"name": "diagnose_performance",
"arguments": {
"app_id": "application_1234567890_0001", // Optional
"time_range_hours": 24 // Optional, default: 24
}
}分析工具
get_spark_logs
获取并分析Spark应用程序日志以进行调试和优化。
{
"name": "get_spark_logs",
"arguments": {
"app_id": "application_1234567890_0001", // Required
"executor_id": "1" // Optional
}
}recommend_configuration
获取Spark和YARN的工作负载特定配置建议。
{
"name": "recommend_configuration",
"arguments": {
"workload_type": "batch", // Options: batch, streaming, ml, interactive
"app_id": "application_1234567890_0001" // Optional
}
}🚀 部署选项
1.EMR引导脚本（推荐）
创建EMR集群时自动部署：
# Upload bootstrap script to S3
aws s3 cp scripts/bootstrap-emr-mcp.sh s3://your-bucket/
# Create EMR cluster with MCP server
aws emr create-cluster \
--name "EMR-MCP-Cluster" \
--release-label emr-6.4.0 \
--applications Name=Spark Name=Hadoop Name=Hive Name=Zeppelin \
--instance-groups \
InstanceGroupType=MASTER,InstanceType=m5.xlarge,InstanceCount=1 \
InstanceGroupType=CORE,InstanceType=m5.2xlarge,InstanceCount=3 \
InstanceGroupType=TASK,InstanceType=m5.large,InstanceCount=2,BidPrice=0.05 \
--bootstrap-actions Path=s3://your-bucket/bootstrap-emr-mcp.sh \
--ec2-attributes KeyName=your-key-pair \
--log-uri s3://your-bucket/emr-logs/2.Docker部署
# Build the image
docker build -t emr-mcp-server .
# Run with docker-compose
docker-compose up -d
# Check logs
docker-compose logs -f emr-mcp-server3.系统化服务
# Copy service file
sudo cp scripts/emr-mcp-server.service /etc/systemd/system/
# Enable and start
sudo systemctl enable emr-mcp-server
sudo systemctl start emr-mcp-server
sudo systemctl status emr-mcp-server💻 使用示例
Python客户端
import asyncio
from examples.client_example import EMRMCPClient
async def main():
async with EMRMCPClient("http://localhost:3000", "emr-mcp-default-key") as client:
# Get cluster information
cluster_info = await client.call_tool("get_cluster_info")
print("Cluster Info:", cluster_info["content"][0]["text"])
# Monitor resources
resources = await client.call_tool("monitor_resources")
print("Resources:", resources["content"][0]["text"])
# Get configuration recommendations
config_rec = await client.call_tool("recommend_configuration", {
"workload_type": "batch"
})
print("Config Recommendations:", config_rec["content"][0]["text"])
asyncio.run(main())cURL示例
# Health check
curl http://localhost:3000/health
# List available tools
curl -X GET http://localhost:3000/tools \
-H "X-API-Key: emr-mcp-default-key"
# Get cluster information
curl -X POST http://localhost:3000/tools/call \
-H "Content-Type: application/json" \
-H "X-API-Key: emr-mcp-default-key" \
-d '{
"name": "get_cluster_info",
"arguments": {}
}'
# Monitor resources
curl -X POST http://localhost:3000/tools/call \
-H "Content-Type: application/json" \
-H "X-API-Key: emr-mcp-default-key" \
-d '{
"name": "monitor_resources",
"arguments": {}
}'🧪 发展
运行测试
# Install development dependencies
pip install -r requirements.txt
# Run all tests
pytest
# Run specific test file
pytest tests/test_cluster.py -v
# Run with coverage
pytest --cov=src tests/ --cov-report=html
# Run demo with mock data
python demo.py
# Test server creation
python test_server.py代码质量
# Format code
black src/ tests/ examples/
# Sort imports
isort src/ tests/ examples/
# Type checking
mypy src/
# Linting
flake8 src/ tests/ examples/🏗️ 建筑
emr-mcp-server/
├── src/
│ ├── server.py # Main MCP server implementation
│ ├── tools/ # MCP tool implementations
│ │ ├── cluster.py # Cluster management tools
│ │ ├── monitoring.py # Resource monitoring tools
│ │ └── analytics.py # Analytics and optimization tools
│ ├── connectors/ # Service connectors
│ │ ├── emr.py # EMR API connector
│ │ ├── yarn.py # YARN ResourceManager connector
│ │ ├── spark.py # Spark History Server connector
│ │ └── hdfs.py # HDFS NameNode connector
│ └── utils/ # Utilities
│ ├── config.py # Configuration management
│ └── auth.py # Authentication utilities
├── config/
│ └── server_config.yaml # Server configuration
├── tests/ # Comprehensive test suite
├── examples/ # Usage examples
├── scripts/ # Deployment scripts
├── Dockerfile # Docker configuration
├── docker-compose.yml # Docker Compose setup
├── demo.py # Demo with mock data
└── test_server.py # Server creation test📊 演示的主要功能
✅ 已完成实施
- 🏗️ 完整的项目结构
- 组织代码库，明确分离关注点 - 带有导入的正确Python包结构 - 使用YAML和环境变量进行配置管理
- 🔧 MCP服务器实现
- 完全符合MCP协议与工具注册 - 异步/等待架构，实现高性能 - 具有可配置格式的结构化日志记录 - 优雅的关机，适当的清理
- 🔌 维修连接器
- 用于集群管理的EMR API集成 - 用于应用程序监控的YARN ResourceManager连接器 - 用于作业分析的Spark历史服务器连接器 - 用于存储监控的HDFS NameNode连接器 - 连接池和重试逻辑
- 🛠️ MCP工具
- 群集管理：get_cluster_info、估计成本、建议实例类型 - 监控：监控资源、分析应用程序、诊断性能 - 分析：get_park_logs，推荐配置 - 所有工具都返回结构化降价，并提供可操作的见解
- 🔒 安全和身份验证
- 多方法身份验证（API密钥、JWT、IAM角色） - 输入验证和净化 - 安全配置管理
- 🚀 部署就绪
- 具有多阶段构建的Docker容器化 - 用于自动部署的EMR引导脚本 - 系统化服务配置 - Docker Compose用于开发
- 🧪 测试与质量
- 带模拟功能的全面测试套件 - 带有真实模拟数据的演示脚本 - 代码质量工具（黑色、isort、mypy、flake8） - 在整个代码库中键入提示
- 📚 文档和示例
- 带有使用示例的详细自述文件 - 带有异步模式的Python客户端示例 - API测试的cURL示例 - 配置示例和部署指南
🎯 演示结果
演示成功显示：
🎯 EMR MCP Server Demo
================================================================================
🚀 EMR Cluster Management Demo
📋 Getting Cluster Information...
💰 Cost Estimation...
🖥️ Instance Type Suggestions...
📊 Resource Monitoring Demo
📈 Resource Monitoring...
🔍 YARN Applications Analysis...
🧠 Analytics & Configuration Demo
⚙️ Configuration Recommendations for Batch Workload...
🤖 Configuration Recommendations for ML Workload...
✅ Demo completed successfully!🔧 生产就绪功能
- 错误处理：通过有意义的消息进行全面的错误处理
- 日志记录：具有多种输出格式的结构化日志记录
- 配置：基于环境的配置和验证
- 监控：健康检查和指标端点
- 安全：身份验证、授权和输入验证
- 演出：异步操作、连接池、缓存
- 部署：多种自动化部署选项
🤝 贡献
我们欢迎捐款！请参阅我们的开发工作流程：
- 分叉 存储库
- 创建 特征分支
- 制造 您通过测试所做的更改
- 跑 测试套件和质量检查
- 提交 pull请求
📄 许可证
此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。
🙏 致谢
- AWS电子病历团队 卓越的大数据平台
- MCP社区 对于协议规范
- Apache Spark 和 哈杜普 社区
______________________________________________________________________
由以下材料制成❤️ EMR社区
*已准备好在EMR集群上进行生产部署！*