spark-sense-ai

Ein MCP-Server (Model Context Protocol), der KI-Agenten – Claude Desktop, Claude Code, Devin oder jeden MCP-kompatiblen Client – zwei Fähigkeiten für die Arbeit mit Apache-Spark-Jobs bietet:

• 🔴 diagnose_spark_failure — ein Spark-Job ist fehlgeschlagen; du erhältst eine Ursachenanalyse und eine konkrete Lösung, basierend auf dem tatsächlichen Fehlerprotokoll und dem spezifischen Code, der fehlgeschlagen ist.

• 🟢 optimize_spark_performance — ein Spark-Job war erfolgreich, ist aber langsam oder teuer; du erhältst gezielte, evidenzbasierte Optimierungsempfehlungen.

Entwickelt von einem Data Engineer mit über 12 Jahren praktischer Apache-Spark-Erfahrung, um denselben Debugging-Instinkt – „Um welche Datei geht es bei diesem Fehler eigentlich und warum?“ – in einen KI-gestützten Workflow zu bringen.

Warum es das gibt

Spark-Fehler lassen sich normalerweise allein aus dem Log diagnostizieren – aber einen 200-zeiligen Stacktrace zu lesen, ihn in einer großen Codebasis mit mehreren Jobs der richtigen Datei zuzuordnen und zu wissen, welche von einem Dutzend möglicher Ursachen tatsächlich vorliegt, erfordert echte Spark-Erfahrung. Dieses Tool automatisiert diesen ersten Schritt: Es findet den relevanten Code (nicht das gesamte Repository), übergibt ihn zusammen mit dem Log an ein LLM und liefert eine strukturierte Diagnose, die du überprüfen und umsetzen kannst.

Was es anders macht

Beispielausgabe

Mit diesem Beispiel-Scala-Fehlerprotokoll und den zugehörigen Projektdateien liefert diagnose_spark_failure (mit konfiguriertem Anbieter) Folgendes zurück:

ROOT CAUSE:
CustomerHelper.validate() calls .trim() on the "email" field without
checking for null first. Records with a missing email cause a
NullPointerException, which aborts the job after 4 failed task retries.

EVIDENCE:
- Caused by: java.lang.NullPointerException: Cannot invoke "String.trim()"
  because "email" is null
- at com.company.jobs.CustomerHelper$.validate(CustomerHelper.scala:22)
- Source shows: email.trim().nonEmpty with no null check beforehand

SUGGESTED FIX:
def validate(row: Row): Boolean = {
  val email = Option(row.getAs[String]("email"))
  email.exists(_.trim.nonEmpty)
}

CONFIDENCE: High

Beachte, dass das Tool automatisch CustomerHelper.scala einbezogen hat (die Datei, auf die das Caused by des Traces tatsächlich zeigt), nicht das gesamte Projekt oder sogar die Top-Level-Einstiegsdatei CustomerOrderJoin.scala – weil der Stacktrace-Parser den tiefsten relevanten Frame aufgelöst hat.

Installation

pip install spark-sense-ai

Installiere Extras nur für das, was du tatsächlich nutzt:

pip install spark-sense-ai[aws]         # for EMR source or Bedrock provider
pip install spark-sense-ai[anthropic]   # for provider="anthropic"
pip install spark-sense-ai[openai]      # for provider="openai"
pip install spark-sense-ai[all]         # everything

provider="none" mit source_type="local" benötigt überhaupt keine Extras – nur die Basis-Abhängigkeit mcp.

Die vier Verwendungsmöglichkeiten

AWS-Anmeldedaten werden bei Bedarf automatisch aus deiner bestehenden Umgebung übernommen (aws configure, eine zugewiesene IAM-Rolle oder Standard-Umgebungsvariablen AWS_*) – niemals als Tool-Parameter übergeben.

Einrichtung

Claude Desktop

Bearbeite claude_desktop_config.json:

{
  "mcpServers": {
    "sparksense": {
      "command": "sparksense-mcp",
      "env": {
        "SPARKSENSE_AWS_REGION": "ap-south-1"
      }
    }
  }
}

Claude Code

claude mcp add sparksense -- sparksense-mcp

Devin

Siehe Devins MCP-Dokumentation für die aktuelle Konfigurationsmethode für deinen Devin-Agentenmodus (Cascade und Devin Local verwenden leicht unterschiedliche Konfigurationsspeicherorte). Richte es genauso wie oben auf den Befehl sparksense-mcp aus.

Anwendungsbeispiele

„Mein Spark-Job ist fehlgeschlagen – EMR-Cluster j-ABC123, Step s-XYZ789. Verwende sparksense, um ihn mit Bedrock zu diagnostizieren.“

„Hier sind das Log meines lokalen Jobs unter ./logs/error.log und der Code unter ./src – diagnostiziere den Fehler.“

„Ich weiß, dass jobs/customer_order_join.py fehlgeschlagen ist – verwende sparksense mit diesem Einstiegspunkt.“

„Verwende sparksense, um das Log unter ./logs/job.log abzurufen – ich werde es selbst überprüfen.“ (provider="none" – das Tool ruft nur ab; der aufrufende Agent übernimmt die Analyse)

„Mein Job war erfolgreich, dauerte aber 40 Minuten. Verwende sparksense, um die Ausführungsstatistiken auf Optimierungsmöglichkeiten zu prüfen.“

Tool-Referenz

diagnose_spark_failure

optimize_spark_performance

Gleiche Parameter wie oben, zusätzlich:

Dateiauswahl-Logik (beide Tools)

1. job_entry_point given?
     → use ONLY that file. No auto-extraction.

2. Else, parse the error log for:
     → Python:      File "<path>", line <N>
     → Scala/Java:  at <package>.<Class>.<method>(<Filename>:<N>)
     (handles mixed PySpark traces — Python frames bottoming into JVM
     frames — by scanning for both patterns in the same log)
     → filters out framework/library internals (site-packages, pyspark,
       org.apache.spark, scala.*, java.*, etc.)
     → fetches up to 10 matched files

3. Else, fallback: broad scan of the project folder, capped at 10 files

Umgebungsvariablen

Tests

git clone https://github.com/YOUR_GITHUB_USERNAME/spark-sense-ai.git
cd spark-sense-ai
pip install -e ".[all]"

# Local source + Anthropic provider, includes Python and Scala samples
export ANTHROPIC_API_KEY="sk-ant-..."
python tests/test_local_anthropic.py

# EMR source + Bedrock provider (needs a real EMR cluster/step)
aws configure
python tests/test_emr_bedrock.py --cluster-id j-XXXXXXX --step-id s-XXXXXXX

Beide Skripte führen zunächst einen kostenlosen Plausibilitätstest ohne API-Aufrufe durch (provider="none"), bevor kostenpflichtige LLM-Aufrufe getätigt werden.

Roadmap

• Automatische Auslösung über Lambda/EventBridge beim Abschluss von EMR-/Glue-Jobs

• Databricks als dritter source_type

• Integration der strukturierten Spark-History-Server-API

• Skew-Erkennung mit Statistiken auf Partitionsebene

Lizenz

MIT – siehe LICENSE.