spark-sense-ai

Un servidor MCP (Model Context Protocol) que ofrece a los agentes de IA — Claude Desktop, Claude Code, Devin, o cualquier cliente compatible con MCP — dos capacidades para trabajar con trabajos de Apache Spark:

• 🔴 diagnose_spark_failure — un trabajo de Spark ha fallado; obtén una causa raíz y una solución concreta, basada en el registro de error real y en el código específico que falló.

• 🟢 optimize_spark_performance — un trabajo de Spark se ha completado pero es lento o costoso; obtén recomendaciones de ajuste específicas y basadas en evidencia.

Creado por un ingeniero de datos con más de 12 años de experiencia práctica con Apache Spark, para llevar ese mismo instinto de depuración — «¿de qué archivo trata realmente este error y por qué?» — a un flujo de trabajo asistido por IA.

Por qué existe esto

Los fallos de Spark suelen poder diagnosticarse solo con el registro, pero leer un stack trace de 200 líneas, hacerlo coincidir con el archivo correcto en un repositorio grande con múltiples trabajos, y saber cuál de una docena de causas posibles es realmente, requiere experiencia real con Spark. Esta herramienta automatiza esa primera pasada: encuentra el código relevante (no todo el repositorio), lo entrega a un LLM junto con el registro, y obtiene un diagnóstico estructurado que puedes verificar y en el que puedes actuar.

Qué lo hace diferente

Ejemplo de salida

Dados este registro de error Scala de ejemplo y sus archivos de proyecto correspondientes, diagnose_spark_failure (con un proveedor configurado) devuelve:

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

Observa que la herramienta incorporó automáticamente CustomerHelper.scala (el archivo al que apunta realmente el Caused by del trace), no todo el proyecto ni siquiera el archivo de entrada de nivel superior CustomerOrderJoin.scala, porque el analizador de stack traces resolvió el marco relevante más profundo.

Instalación

pip install spark-sense-ai

Instala solo los extras que vayas a usar realmente:

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" con source_type="local" no necesita ningún extra — solo la dependencia base mcp.

Las cuatro formas de usarlo

Las credenciales de AWS, cuando son necesarias, se toman automáticamente de tu configuración existente (aws configure, un rol IAM adjunto o las variables de entorno AWS_* estándar) — nunca se pasan como parámetro de la herramienta.

Configuración

Claude Desktop

Edita 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

Consulta la documentación de MCP de Devin para conocer el método de configuración actual según tu modo de agente de Devin (Cascade y Devin Local usan ubicaciones de configuración ligeramente diferentes). Configúralo para que use el comando sparksense-mcp de la misma manera que arriba.

Ejemplos de uso

"Mi trabajo de Spark ha fallado — clúster EMR j-ABC123, paso s-XYZ789. Usa sparksense para diagnosticarlo con Bedrock."

"Aquí está el registro de mi trabajo local en ./logs/error.log y el código en ./src — diagnostica el fallo."

"Sé que es jobs/customer_order_join.py el que ha fallado — usa sparksense con eso como punto de entrada."

"Usa sparksense para obtener el registro en ./logs/job.log — lo revisaré yo mismo." (provider="none" — la herramienta solo obtiene; el agente que llama hace el razonamiento)

"Mi trabajo se completó, pero tardó 40 minutos. Usa sparksense para revisar las estadísticas de ejecución en busca de oportunidades de optimización."

Referencia de la herramienta

diagnose_spark_failure

optimize_spark_performance

Mismos parámetros que arriba, más:

Lógica de selección de archivos (ambas herramientas)

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

Variables de entorno

Pruebas

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

Ambos scripts ejecutan primero una comprobación de cordura gratuita, sin llamadas a API (provider="none"), antes de realizar cualquier llamada facturable a un LLM.

Hoja de ruta

• Disparo automático mediante Lambda/EventBridge al completarse trabajos de EMR/Glue

• Databricks como tercer source_type

• Integración estructurada con la API de Spark History Server

• Detección de sesgo (skew) con estadísticas a nivel de partición

Licencia

MIT — consulta LICENSE.