PySpark MCP

PySpark MCP Server

SQL migration assistance, AWS Glue job template generation, and Spark code optimization — as an MCP server.

Not the live-Spark pyspark-mcp package. This project is SQL → PySpark / Glue source generation, published as pyspark-tools. SemyonSinchenko/pyspark-mcp introspects a running SparkSession. A deprecated pyspark-mcp console script remains here so old configs keep working; it prints a warning, then starts this server.

[CI Pipeline] [Python 3.11+] [License: MIT]

What It Does

• SQL Dialect Transpilation — Convert between PostgreSQL, Oracle, Redshift, MySQL, Snowflake, and Spark SQL using SQLGlot
• PySpark DataFrame API Generation — Generate DataFrame API source text from SQL, with optimization hints
• AWS Glue templates — Job script strings, DynamicFrame conversions, Data Catalog definitions, S3 layout advice
• Batch Processing — Walk SQL files/directories and emit converted modules
• Code Review & Optimization — Pattern-based review of existing PySpark source
• Pattern Detection — Find duplicated snippets and suggest utilities

What It Doesn't Do

• Recursive CTEs → provides Spark SQL equivalent + guidance (PySpark has no native recursive CTE support)
• MERGE/PIVOT/CONNECT BY → transpiles to Spark SQL, provides DataFrame API guidance
• Perfect 1:1 DataFrame API transpilation for all SQL — complex queries get Spark SQL + recommendations
• It does not start a SparkSession, submit Glue jobs, or execute SQL
• optimize(mode="code") returns suggestions; it does not rewrite your code
• glue_s3 is a path heuristic (no AWS call, no measured speedups)
• It does not replace SemyonSinchenko/pyspark-mcp for live catalog/plans

Why this vs calling sqlglot yourself

SQLGlot already transpiles dialects. This MCP adds three things around that kernel: DataFrame-API pretty-printing with join/window/cast mappings that the conversion tests lock, Glue job boilerplate strings (bookmarks, DynamicFrames, catalog tables) so an agent can emit a file instead of assembling one, and a 14-tool FastMCP surface so an LLM picks convert / mode=sql instead of wiring sqlglot itself. If you only need sqlglot.transpile(...), use sqlglot.

Quick Start

pip install pyspark-toolspyspark-tools

Zero-clone alternative: uvx pyspark-tools. run_server.py is a development convenience that inserts sys.path and prints startup banners. Prefer pyspark-tools in configs and production.

Try it

pip install pyspark-toolspython -c "from pathlib import Path; from pyspark_tools.sql_converter import SQLToPySparkConverter as C; from pyspark_tools.consolidated_tools import glue_job; c,s,o=C(),Path('examples'),Path('examples/out'); [(o/f'{n}.py').write_text(c.convert_sql_to_pyspark((s/f'{n}.sql').read_text(), dialect=d).pyspark_code) for n,d in [('postgres_orders','postgres'),('oracle_decode','oracle')]]; (o/'orders_etl_glue.py').write_text(glue_job(mode='template', job_name='orders_etl', sql_query=(s/'postgres_orders.sql').read_text())['template'])"

Writes the same files as examples/out/. MCP stdio CLI: pyspark-tools.

Example: SQL → PySpark

SELECT o.customer_id, c.name, SUM(o.amount) AS totalFROM orders oJOIN customers c ON o.customer_id = c.idWHERE o.status = 'paid'GROUP BY o.customer_id, c.name

Call convert with mode=sql. Captured converter output (dialect=spark):

from pyspark.sql import SparkSessionfrom pyspark.sql.functions import (    col, lit, when, count, sum as spark_sum, avg, min, max, countDistinct,    coalesce, concat, datediff, date_add, to_date,    row_number, rank, lag, lead,)from pyspark.sql.window import Window
# Generated from SPARK SQLspark = SparkSession.builder.appName('SQLToPySpark').getOrCreate()
# Load table: customerscustomers_df = spark.table('customers')# Load table: ordersorders_df = spark.table('orders')
# Main queryresult_df = (orders_df.alias('o')    .join(customers_df.alias('c'), (col('o.customer_id') == col('c.id')), 'inner')    .filter((col('o.status') == lit('paid')))    .groupBy(col('o.customer_id'), col('c.name'))    .select(col('o.customer_id'), col('c.name'), (spark_sum(col('o.amount'))).alias('total')))

Exact output depends on dialect detection and fallbacks; conversion tests in tests/test_sql_conversion_fixes.py pin the important constructs. Notebook-style import * / show() is opt-in via style="notebook" on the converter.

MCP Configuration

Claude Desktop

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

Linux: ~/.config/Claude/claude_desktop_config.json

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

Hermes Agent

Add to ~/.hermes/config.yaml:

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

Docker

The image is stdio only (FastMCP over stdin/stdout). There is no HTTP server on port 8000. docker compose up is for local tests, not a health-checkable web service.

docker compose --profile test run --rm pyspark-tools-test

Tools

Three primary tools. The other eleven routers stay registered this minor version but are deprecated — prefer convert, glue_job, and review.

convert — SQL → PySpark (including mode=batch_dir)

convert(mode="sql", sql_query="SELECT id FROM users", dialect="postgres")convert(mode="batch_dir", directory_path="etl/", output_dir="out")

glue_job — Glue 5.0 job template strings

glue_job(mode="template", job_name="orders_etl", sql_query="SELECT * FROM orders")

review — code review, patterns, duplicates

review(mode="code", code="df = spark.table('t')\ndf.collect()")

Legacy / deprecated: analyze, optimize, glue_schema, glue_s3, glue_data, refactor, search, context, batch_status, s3_source, analytics. Still callable; do not advertise to new agents.

Security

This MCP can read local files (SQL, TXT, PDF) and, if the [aws] extra is installed, list/read S3 with the host's default AWS credentials. File tools only allow paths under the process working directory (or an explicit base_path / FileHandler(base_directory=...)). That is not a sandbox.

Run the server under a restricted OS account. Do not point it at secrets directories. Do not attach AWS credentials with write access unless you intend S3 reads via s3_source / glue_s3. Optional extras:

pip install "pyspark-tools[aws]"    # boto3 for S3/Glue catalog helperspip install "pyspark-tools[spark]"  # pyspark — not required at runtime; generated code only

Development

python -m venv .venvsource .venv/bin/activatepip install -e ".[dev]"
# Testpytest tests/ -v --cov=pyspark_tools
# Formatblack pyspark_tools testsisort pyspark_tools tests
# Lintflake8 pyspark_tools tests

Requires Python 3.11+ (matches the CI matrix).

Architecture

pyspark_tools/├── server.py              # FastMCP server + helper implementations├── consolidated_tools.py  # 14 @app.tool() routers├── 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 (optional boto3)└── file_utils.py          # File I/O with allow-root checks

License

MIT — see LICENSE.

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