Spark MCP Server

Spark历史服务器MCP代理

A. 基于stdio的MCP服务器 通过以下方式公开Spark历史服务器数据 4个标准化MCP工具此服务器使用模型上下文协议将MCP客户端直接连接到Spark历史服务器REST API。

🚀 它做什么

提供 4个MCP工具 查询您的Spark历史服务器：

1. 获取应用程序 -列出具有可选筛选功能的Spark应用程序
2. 获取应用程序信息 -获取详细的应用程序信息
3. get_application_jobs -为特定应用程序获取工作
4. get_application_stages -获取特定应用程序的阶段

🏗️ 建筑

┌─────────────────┐    stdio/MCP    ┌──────────────────┐    HTTP     ┌──────────────────┐
│   MCP Client    │ ──────────────── │   MCP Server     │ ─────────── │ Spark History    │
│                 │                  │   (This Project) │             │ Server :18080    │
└─────────────────┘                  └──────────────────┘             └──────────────────┘

重要：这使用基于stdio的MCP协议（不是HTTP端点），意思是：

• 通过stdin/stdout而非网络端口进行通信
• 使用JSON-RPC 2.0进行消息格式化
• 需要官方 mcp>=1.13.0 包裹

✨ 特点

• ✅ 标准MCP协议：使用官方MCP SDK和stdio传输
• ✅ 直接历史服务器访问：Spark历史服务器REST API的纯代理
• ✅ 4核心工具：基本应用程序、作业和阶段数据访问
• ✅ 无状态操作：不需要本地存储或数据持久性
• ✅ 简单配置：只需要一个历史服务器URL

🛠️ 可用的MCP工具

🚧 尚未实施

这些工具计划在未来实施：

💻 快速开始

先决条件

• Spark历史服务器正在运行并可访问（通常在端口18080上）
• Python 3.8+
• MCP兼容客户端

安装

1. 克隆存储库：

   git clone 
   cd spark-mcp-server

1. 安装依赖项：

   # Dependencies are already installed in the virtual environment
   # If you need to reinstall:
   ./env/bin/python3 -m pip install -r requirements.txt

1. 启动Spark历史服务器：

   ./start_history_server.sh

这将启动历史服务器http://localhost:18080

⚠️ 重要提示-事件日志目录： Spark历史服务器从以下位置读取事件日志 /tmp/spark-events/ 目录。让服务器显示数据：

• 所有Spark应用程序都必须将事件日志写入此目录
• 集 spark.eventLog.dir=/tmp/spark-events 在Spark配置中
• 或者使用环境变量： export SPARK_EVENTLOG_DIR=/tmp/spark-events
• 确保此目录存在并且可访问

MCP客户端设置

配置您的MCP客户端以使用此服务器。此设置对于正确的MCP stdio协议通信至关重要：

"spark-history-server": {
  "command": "/path/to/spark-mcp-server/env/bin/python3",
  "source": "custom",
  "args": [
    "/path/to/spark-mcp-server/src/main.py",
    "--config",
    "/path/to/spark-mcp-server/config.json"
  ],
  "env": {
    "PYTHONPATH": "/path/to/spark-mcp-server"
  }
}

⚠️ 重要:

• 更新路径以匹配您的实际安装目录
• 服务器使用stdio协议，而不是HTTP端点
• 需要安装MCP SDK（MCP>=1.13.0）

🧪 测试

生成示例数据

1. 运行示例Spark应用程序：

   python3 test-files/sample_spark_app.py

备注：示例应用程序配置为将事件日志写入 /tmp/spark-events/ 其与历史服务器配置匹配。

1. 启动历史服务器：

   ./start_history_server.sh

1. 验证数据是否可用：

   curl "http://localhost:18080/api/v1/applications?limit=3"

💡 对于您自己的Spark应用程序： 要使Spark应用程序在历史服务器中可见，请确保它们将事件日志写入同一目录：

# Using spark-submit
spark-submit \
  --conf spark.eventLog.enabled=true \
  --conf spark.eventLog.dir=/tmp/spark-events \
  your_app.py

# Using environment variable
export SPARK_EVENTLOG_DIR=/tmp/spark-events
spark-submit --conf spark.eventLog.enabled=true your_app.py

# In PySpark code
spark = SparkSession.builder \
  .config("spark.eventLog.enabled", "true") \
  .config("spark.eventLog.dir", "/tmp/spark-events") \
  .getOrCreate()

测试MCP服务器

# Test version
./env/bin/python3 src/main.py --version

# Test MCP protocol (initialize message)
echo '{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test", "version": "1.0"}}}' | \
./env/bin/python3 src/main.py --config config.json

# Test tools listing
echo '{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}' | \
./env/bin/python3 src/main.py --config config.json

📊 响应示例

应用程序列表

{
  "success": true,
  "data": [
    {
      "id": "local-1755324061532",
      "name": "MCP-Test-Sample-Application",
      "attempts": [{
        "startTime": "2025-08-16T06:01:01.024GMT",
        "endTime": "2025-08-16T06:01:45.732GMT",
        "completed": true,
        "sparkUser": "username",
        "appSparkVersion": "3.3.1"
      }]
    }
  ],
  "count": 1,
  "message": "Retrieved 1 applications"
}

应用程序作业

{
  "success": true,
  "data": [
    {
      "jobId": 0,
      "name": "count at NativeMethodAccessorImpl.java:0",
      "status": "SUCCEEDED",
      "numTasks": 8,
      "numCompletedTasks": 8,
      "submissionTime": "2025-08-16T06:01:03.732GMT",
      "completionTime": "2025-08-16T06:01:04.752GMT"
    }
  ],
  "count": 34,
  "message": "Retrieved 34 jobs for application local-1755324061532"
}

📁 项目结构

spark-mcp-server/
├── src/
│   ├── main.py              # MCP server entry point (stdio-based)
│   ├── history_client.py    # Spark History Server HTTP client
│   └── mcp_server.py       # Original implementation (unused)
├── config.json             # Server configuration
├── start_history_server.sh # History Server startup script
├── requirements.txt        # Python dependencies
├── env/                    # Virtual environment (with MCP SDK)
├── test-files/            # Development & test files
└── README.md              # This file

⚙️ 配置选项

基本配置

{
  "spark_history_server": {
    "url": "http://localhost:18080"
  },
  "logging": {
    "level": "INFO",
    "console": true
  }
}

通过身份验证

{
  "spark_history_server": {
    "url": "https://spark-history.company.com:18080",
    "auth": {
      "type": "basic",
      "username": "spark_user",
      "password": "${SPARK_PASSWORD}"
    }
  },
  "logging": {
    "level": "INFO",
    "console": true,
    "file": "/var/log/spark-mcp.log"
  }
}

🎯 用例

• 性能分析：查询作业/阶段执行时间和资源使用情况
• 监控集成：将Spark指标输入监控仪表板
• 开发工具：用于Spark应用程序监控的IDE集成
• CI/CD管道：自动Spark作业状态检查
• 数据工程：对Spark执行元数据的编程访问

🚧 当前限制

• 有限的工具：计划的15+个工具中只有4个得到实施
• 基本错误处理：最小的错误处理和重试逻辑
• 基本身份验证：已实施，但需要额外测试
• 单一历史服务器：不支持多个历史服务器实例
• 需要安装：客户端必须配置正确的路径

📜 许可证

Apache许可证2.0-有关详细信息，请参阅许可证文件。

🐛 支持

• 检查 test-files/ 示例和故障排除目录
• 使用运行时查看日志 "level": "DEBUG" 在配置中
• 确保Spark历史服务器可以在配置的URL上访问
• 看 MCP_CONFIGURATION_FIX.md 有关详细的故障排除信息

______________________________________________________________________

当前状态: ✅ 工作MCP服务器 实施了4个核心工具。