pyspark MCP

PySpark MCP服务器

SQL迁移协助、AWS Glue作业生成和Spark代码优化——作为MCP服务器。

  

它做什么

• SQL方言转换 --使用以下命令在PostgreSQL、Oracle、Redshift、MySQL、Snowflake和Spark SQL之间进行转换 SQLGlot
• PySpark DataFrame API生成 -使用优化提示从SQL生成DataFrame API代码
• AWS胶水集成 --作业模板、DynamicFrame转换、数据目录定义、S3优化策略
• 批处理 --同时处理数百个SQL文件
• 代码审查和优化 --分析现有的PySpark代码以提高性能
• 模式检测 --查找代码重复并建议重构

它不做什么

• 递归CTE→ 提供Spark SQL等效+指导（PySpark没有本机递归CTE支持）
• 合并/枢轴/连接方式→ 可移植到Spark SQL，提供DataFrame API指南
• 完美的1:1 DataFrame API转换所有SQL-复杂查询获得Spark SQL+优化建议

快速开始

pip install -e .
pyspark-mcp  # starts the MCP server

MCP配置

克劳德桌面版

添加 ~/Library/Application Support/Claude/claude_desktop_config.json:

{
  "mcpServers": {
    "pyspark": {
      "command": "pyspark-mcp",
      "args": []
    }
  }
}

赫尔墨斯代理

添加 ~/.hermes/config.yaml:

mcp:
  servers:
    pyspark:
      command: pyspark-mcp
      enabled_tools: all

码头工人

docker compose up -d

工具

SQL转换

• convert_sql_to_pyspark --使用方言检测将SQL转换为PySpark
• analyze_sql_context --分析SQL复杂性并提出方法

AWS胶水

• generate_aws_glue_job_template --生成完整的胶水作业脚本
• convert_dataframe_to_dynamic_frame --DataFrame↔ 动态帧转换
• generate_data_catalog_table_definition --数据目录表定义
• generate_incremental_processing_job --增量/CDC工作生成
• analyze_s3_optimization_opportunities --S3布局和分区分析

优化

• review_pyspark_code --代码审查及性能建议
• optimize_pyspark_code --建议对现有代码进行优化
• recommend_join_strategy --广播与随机加入推荐
• suggest_partitioning_strategy --分区建议

批处理

• batch_process_files --同时处理多个SQL文件
• batch_process_directory --转换整个目录

发展

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

# Test
pytest tests/ -v --cov=pyspark_tools

# Format
black pyspark_tools tests
isort pyspark_tools tests

# Lint
flake8 pyspark_tools tests

建筑

pyspark_tools/
├── server.py              # FastMCP server + tool definitions
├── sql_converter.py       # SQLGlot-based transpilation + DataFrame API generation
├── aws_glue_integration.py # Glue job templates, DynamicFrame, Data Catalog
├── advanced_optimizer.py  # Performance analysis + optimization suggestions
├── batch_processor.py     # Concurrent file processing
├── code_reviewer.py       # PySpark code review patterns
├── duplicate_detector.py  # Code deduplication
├── data_source_analyzer.py # Data source analysis
└── file_utils.py          # File I/O utilities

CI/CD

• ✅ 256项测试通过
• ✅ 71%的代码覆盖率
• ✅ 代码质量检查（黑色、isort、flake8）
• ✅ Python 3.11测试

许可证

麻省理工学院——见 许可证.

______________________________________________________________________

mcp-name: io.github.AnnasMazhar/pyspark-mcp