使用说明

Spark历史服务器MCP代理

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

🚀 它做什么

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

获取应用程序 -列出具有可选筛选功能的Spark应用程序
获取应用程序信息 -获取详细的应用程序信息
get_application_jobs -为特定应用程序获取工作
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工具

工具	描述	参数	状态
`get_applications`	列出所有应用程序	`status` （可选）， `limit` （可选）	✅ 实现
`get_application_info`	获取应用程序详细信息	`app_id` （必填）	✅ 实现
`get_application_jobs`	获取应用程序作业	`app_id` （必填）	✅ 实现
`get_application_stages`	获取申请阶段	`app_id` （必填）， `status` （可选）	✅ 实现

🚧 尚未实施

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

工具	描述	状态
`get_application_executors`	获取执行人信息	🚧 计划的
`get_application_environment`	获取环境详细信息	🚧 计划的
`get_job_info`	获取具体的工作详细信息	🚧 计划的
`get_stage_info`	获取特定阶段的详细信息	🚧 计划的
`get_stage_tasks`	获取阶段任务详细信息	🚧 计划的
`get_rdd_storage`	获取RDD存储信息	🚧 计划的
`get_sql_queries`	获取SQL执行数据	🚧 计划的
`get_streaming_batches`	获取流式批处理数据	🚧 计划的

💻 快速开始

先决条件

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

安装

克隆存储库：

git clone <repository-url>
cd spark-mcp-server

安装依赖项：

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

启动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）

🧪 测试

生成示例数据

运行示例Spark应用程序：
```
python3 test-files/sample_spark_app.py
```
备注：示例应用程序配置为将事件日志写入 /tmp/spark-events/ 其与历史服务器配置匹配。
启动历史服务器：
```
./start_history_server.sh
```

验证数据是否可用：

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个核心工具。