spark-sense-ai

MCP-сервер (Model Context Protocol), который предоставляет AI-агентам — Claude Desktop, Claude Code, Devin или любому MCP-совместимому клиенту — две возможности для работы с задачами Apache Spark:

• 🔴 diagnose_spark_failure — задача Spark завершилась с ошибкой; получите первопричину и конкретное исправление, основанное на фактическом журнале ошибок и конкретном коде, который не сработал.

• 🟢 optimize_spark_performance — задача Spark выполнилась, но работает медленно или требует больших затрат; получите адресные рекомендации по настройке, основанные на фактах.

Создан инженером данных с 12+ годами практического опыта работы с Apache Spark, чтобы перенести тот самый инстинкт отладчика — «о каком именно файле на самом деле говорит эта ошибка и почему» — в процесс с использованием ИИ.

Зачем это нужно

Сбои Spark обычно можно диагностировать по одному лишь журналу, но чтение стек-трейса на 200 строк, сопоставление его с нужным файлом в большой кодовой базе с множеством задач и понимание, какая из дюжины возможных причин действительно имеет место, требует реального опыта работы со Spark. Этот инструмент автоматизирует этот первый этап: он находит релевантный код (а не весь репозиторий), передаёт его LLM вместе с журналом и возвращает структурированный диагноз, который можно проверить и на основе которого можно действовать.

Чем он отличается

Пример вывода

На основе этого примера журнала ошибок Scala и его соответствующих файлов проекта функция diagnose_spark_failure (при настроенном провайдере) возвращает:

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

Обратите внимание: инструмент автоматически подтянул CustomerHelper.scala (файл, на который на самом деле указывает Caused by из трейса), а не весь проект и даже не корневой входной файл CustomerOrderJoin.scala — потому что парсер стек-трейса определил самый глубокий значимый фрейм.

Установка

pip install spark-sense-ai

Устанавливайте дополнительные зависимости только под то, что реально будете использовать:

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" с source_type="local" вообще не требует дополнительных зависимостей — только базовую зависимость mcp.

Четыре способа использования

Учётные данные AWS при необходимости подхватываются автоматически из вашего существующего окружения (aws configure, назначенная IAM-роль или стандартные переменные окружения AWS_*) — они никогда не передаются как параметр инструмента.

Настройка

Claude Desktop

Отредактируйте 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

См. документацию Devin по MCP — там описан актуальный способ настройки для вашего режима агента Devin (Cascade и Devin Local используют немного разные расположения конфигурации). Укажите команду sparksense-mcp так же, как показано выше.

Примеры использования

«Моя задача Spark завершилась ошибкой — кластер EMR j-ABC123, шаг s-XYZ789. Используй sparksense, чтобы диагностировать её с помощью Bedrock.»

«Вот журнал моей локальной задачи в ./logs/error.log и код в ./src — диагностируй сбой.»

«Я знаю, что упал именно jobs/customer_order_join.py — используй sparksense с этим в качестве точки входа.»

«Используй sparksense, чтобы получить журнал из ./logs/job.log — я сам его изучу.» (provider="none" — инструмент только получает данные; анализ выполняет вызывающий агент)

«Моя задача выполнилась, но заняла 40 минут. Используй sparksense, чтобы проверить статистику выполнения на предмет возможностей оптимизации.»

Справочник инструментов

diagnose_spark_failure

optimize_spark_performance

Те же параметры, что и выше, плюс:

Логика выбора файлов (для обоих инструментов)

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

Переменные окружения

Тестирование

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

Оба скрипта сначала выполняют бесплатную проверку без вызовов API (provider="none"), прежде чем совершать какие-либо платные вызовы LLM.

Дорожная карта

• Автоматический запуск через Lambda/EventBridge по завершении задач EMR/Glue

• Databricks как третий source_type

• Интеграция со структурированным API Spark History Server

• Обнаружение перекоса данных с помощью статистики на уровне партиций

Лицензия

MIT — см. LICENSE.