PySpark MCP

PySpark MCP Server

SQL 移行支援、AWS Glue ジョブのテンプレート生成、Spark コードの最適化を MCP サーバーとして提供します。

ライブ Spark を扱う pyspark-mcp パッケージではありません。 このプロジェクトは SQL → PySpark／Glue のソース生成を行い、pyspark-tools として公開されています。 SemyonSinchenko/pyspark-mcp は、実行中の SparkSession をイントロスペクションします。非推奨の pyspark-mcp コンソールスクリプトも、古い設定との互換性を保つために残されています。警告を表示した後、このサーバーを起動します。

できること

• SQL 方言の変換 — SQLGlot を使って、PostgreSQL、Oracle、Redshift、MySQL、Snowflake、Spark SQL の間で変換します
• PySpark DataFrame API の生成 — 最適化のヒントを含め、SQL から DataFrame API のソーステキストを生成します
• AWS Glue テンプレート — ジョブスクリプト文字列、DynamicFrame 変換、Data Catalog 定義、S3 レイアウトのアドバイスを提供します
• バッチ処理 — SQL ファイル／ディレクトリを走査し、変換済みモジュールを出力します
• コードレビューと最適化 — 既存の PySpark ソースをパターンに基づいてレビューします
• パターン検出 — 重複したスニペットを見つけ、ユーティリティを提案します

できないこと

• 再帰 CTE → Spark SQL の同等構文とガイダンスを提供します（PySpark はネイティブの再帰 CTE に対応していません）
• MERGE／PIVOT／CONNECT BY → Spark SQL に変換し、DataFrame API のガイダンスを提供します
• あらゆる SQL に対する完全な 1:1 の DataFrame API 変換 — 複雑なクエリには Spark SQL と推奨事項を提示します
• SparkSession の開始、Glue ジョブの送信、SQL の実行は行いません
• optimize(mode="code") が返すのは提案であり、コードを書き換えることはありません
• glue_s3 はパスのヒューリスティックです（AWS 呼び出しや、測定済みの高速化はありません）
• ライブのカタログ／実行プラン向けの SemyonSinchenko/pyspark-mcp を置き換えるものではありません

自分で sqlglot を呼び出す場合との違い

SQLGlot はすでに方言を変換できます。この MCP は、その中核機能に 3 つの要素を追加します。変換テストで仕様を固定した、結合／ウィンドウ／キャストのマッピングを含む DataFrame API の整形出力、エージェントがファイルを組み立てずに出力できるようにする Glue ジョブの定型文字列（ブックマーク、DynamicFrame、カタログテーブル）、そして LLM が sqlglot 自体を組み込むのではなく、convert／mode=sql を選べる 14 ツールの FastMCP インターフェースです。sqlglot.transpile(...) だけが必要な場合は、sqlglot をお使いください。

クイックスタート

pip install pyspark-tools
pyspark-tools

リポジトリをクローンしない方法：uvx pyspark-tools。run_server.py は sys.path を挿入して起動バナーを表示する、開発用の便利なスクリプトです。設定や本番環境では pyspark-tools をお使いください。

お試しください

pip install pyspark-tools
python -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'])"

examples/out/ と同じファイルを書き込みます。MCP stdio CLI：pyspark-tools。

例：SQL → PySpark

SELECT o.customer_id, c.name, SUM(o.amount) AS total
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.status = 'paid'
GROUP BY o.customer_id, c.name

mode=sql を指定して convert を呼び出します。変換器の出力例（dialect=spark）：

from pyspark.sql import SparkSession
from 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 SQL
spark = SparkSession.builder.appName('SQLToPySpark').getOrCreate()

# Load table: customers
customers_df = spark.table('customers')
# Load table: orders
orders_df = spark.table('orders')

# Main query
result_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')))

正確な出力は方言の検出とフォールバックによって異なります。重要な構文は、tests/test_sql_conversion_fixes.py の変換テストで固定されています。ノートブック形式の import *／show() は、変換器で style="notebook" を指定した場合にのみ有効になります。

MCP の設定

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

~/.hermes/config.yaml に追加します：

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

Docker

このイメージはstdio 専用です（stdin／stdout 経由の FastMCP）。ポート 8000 で HTTP サーバーが起動することはありません。docker compose up はローカルテスト用であり、ヘルスチェック可能な Web サービスではありません。

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

ツール

主要なツールは3つです。残りの11個のルーターはこのマイナーバージョンでも登録されたままですが、非推奨です。convert、glue_job、review を優先してください。

convert — SQL → PySpark（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 ジョブの テンプレート 文字列

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

review — コードレビュー、パターン、重複

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

レガシー／非推奨： analyze、optimize、glue_schema、glue_s3、 glue_data、refactor、search、context、batch_status、s3_source、 analytics。引き続き呼び出せますが、新しいエージェントには案内しないでください。

セキュリティ

この MCP はローカルファイル（SQL、TXT、PDF）を読み取ることができ、[aws] エクストラがインストールされている場合は、ホストのデフォルト AWS 認証情報を使用して S3 の一覧表示と読み取りも行えます。ファイルツールで許可されるパスは、プロセスの作業ディレクトリ配下（または明示的に指定した base_path ／ FileHandler(base_directory=...)）のみです。これはサンドボックスではありません。

サーバーは、制限された OS アカウントで実行してください。シークレットのディレクトリを参照させないでください。s3_source ／ glue_s3 経由で S3 を読み取る場合を除き、書き込みアクセス権のある AWS 認証情報を設定しないでください。オプションのエクストラ：

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

開発

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

Python 3.11 以降が必要です（CI マトリックスに合わせています）。

アーキテクチャ

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

ライセンス

MIT — LICENSE を参照してください。

---

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