spark-submit: упаковка зависимостей и запуск на кластере

Полный разбор spark-submit: анатомия команды, упаковка Python и JVM зависимостей, client vs cluster mode, управление ресурсами, диагностика проблем запуска и production best practices.

core internals

spark-submit - единственный правильный способ запустить Spark-приложение на кластере. Попытка запустить python main.py или java -jar app.jar напрямую создаст Driver-процесс без Executor-ов - Spark просто не узнает о кластере. spark-submit запускает Spark Runtime, инициализирует связь с кластер-менеджером и гарантирует, что весь код доедет до каждого Executor.

Зачем нужен spark-submit¶

Когда вы пишете spark = SparkSession.builder.getOrCreate() в PySpark, за этим стоит предположение: Spark Runtime уже запущен и настроен. В локальном режиме это прозрачно. На кластере нужно:

Указать, кто управляет ресурсами (YARN, Kubernetes, Standalone)
Доставить код приложения на каждый Executor
Доставить зависимости (Python-пакеты, JAR-файлы)
Сказать, сколько памяти и CPU выделить

Всё это делает spark-submit. Он - артефакт запуска, а не просто обёртка.

Анатомия команды spark-submit¶

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name "ETL Pipeline v2" \
  --driver-memory 4g \
  --driver-cores 2 \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 10 \
  --conf spark.sql.shuffle.partitions=400 \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --py-files deps.zip \
  --files config/production.yaml \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0 \
  main.py \
  --date 2024-01-15 \
  --env production

Аргументы после имени скрипта (main.py) передаются в приложение - читаются через sys.argv или argparse.

Ключевые флаги¶

Флаг	Описание
`--master`	Кластер-менеджер: `yarn`, `k8s://...`, `spark://host:7077`, `local[*]`
`--deploy-mode`	`client` или `cluster` - где живёт Driver
`--name`	Имя приложения в Spark UI и YARN RM
`--driver-memory`	Память JVM Driver (+ overhead ~10%)
`--driver-cores`	CPU для Driver (только Standalone и K8s)
`--executor-memory`	Память JVM каждого Executor
`--executor-cores`	vCPU каждого Executor
`--num-executors`	Кол-во Executor (отключает Dynamic Allocation)
`--conf key=val`	Переопределение spark-defaults.conf «на лету»
`--py-files`	Python: `.py`, `.zip`, `.egg` для Executor-ов
`--files`	Файлы в рабочий каталог каждого Executor
`--archives`	Архивы: `.tar.gz`, `.zip` - распаковываются на Executor
`--jars`	JAR-файлы в classpath Driver и Executor
`--packages`	Maven-координаты: скачиваются из Maven Central
`--exclude-packages`	Исключить пакеты из `--packages`
`--repositories`	Дополнительные Maven-репозитории
`--class`	Точка входа для JVM-приложений (Scala/Java)
`--queue`	YARN-очередь (только YARN)
`--principal` / `--keytab`	Kerberos-аутентификация

Deploy mode: client vs cluster¶

Характеристика	`--deploy-mode client`	`--deploy-mode cluster`
Где живёт Driver	Машина запуска (edge node, ноутбук)	Контейнер внутри кластера
Если закрыть терминал	Job умирает	Job продолжается
Сетевой трафик	Через клиентскую сеть	Только внутри кластера
Логи Driver	Прямо в терминале	`yarn logs -applicationId ...`
Spark UI	`localhost:4040`	Адрес в RM UI → Application Master
Когда использовать	Отладка, разработка, Jupyter	Production, Airflow, cron
Ограничение	Edge node должен иметь сеть до YARN	Код должен быть доступен из кластера

В cluster mode скрипт main.py должен быть доступен YARN NodeManager-у - нельзя передать локальный файл ~/projects/main.py. Нужен путь hdfs://... или предварительный spark-submit распакует артефакт через Staging Area.

Как Spark доставляет код: Staging Area¶

Staging Area - временная директория в HDFS (/user/{user}/.sparkStaging/{appId}/). Spark автоматически создаёт её при запуске и удаляет после завершения. Все --py-files, --files, --jars сначала копируются туда, и только потом раздаются Executor-ам.

Упаковка Python-зависимостей¶

--py-files: базовый вариант¶

# Один файл
spark-submit --py-files utils.py main.py

# ZIP с несколькими модулями
zip deps.zip src/transforms.py src/validators.py src/connectors.py
spark-submit --py-files deps.zip main.py

# Несколько источников через запятую
spark-submit --py-files utils.py,helpers.zip,extra.egg main.py

Файлы из --py-files добавляются в sys.path каждого Executor. Они не устанавливаются как пакеты - просто распаковываются в рабочий каталог.

ZIP-архив модуля¶

project/
├── main.py
└── myapp/
    ├── __init__.py
    ├── etl.py
    └── utils.py

cd project
zip -r myapp.zip myapp/
spark-submit --py-files myapp.zip main.py

В main.py работает как обычно:

from myapp.etl import transform
from myapp.utils import clean

--archives: передача виртуального окружения¶

Самый надёжный способ - упаковать всё окружение в архив и распаковать на каждом Executor:

# Создать conda environment
conda create -n spark-env python=3.10 pandas=2.1 pyarrow=14.0
conda activate spark-env
conda pack -o spark-env.tar.gz

# Передать и использовать
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --archives spark-env.tar.gz#spark-env \
  --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./spark-env/bin/python \
  --conf spark.executorEnv.PYSPARK_PYTHON=./spark-env/bin/python \
  main.py

#spark-env - псевдоним директории после распаковки. conda-pack создаёт самодостаточный архив без внешних зависимостей от системы.

То же самое с venv:

python -m venv spark-env
source spark-env/bin/activate
pip install pandas pyarrow delta-spark

# Упаковать venv в zip (conda-pack предпочтительнее)
cd spark-env && zip -r ../spark-env.zip . && cd ..

spark-submit \
  --archives spark-env.zip#spark-env \
  --conf spark.executorEnv.PYSPARK_PYTHON=./spark-env/bin/python \
  main.py

Упаковка JVM-зависимостей (Scala/Java)¶

Fat JAR (Uber JAR)¶

Стратегия: упаковать все зависимости в один JAR. Executor-ы получают один файл - никаких проблем с classpath.

sbt + sbt-assembly:

// project/plugins.sbt
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.5")

// build.sbt
assembly / assemblyMergeStrategy := {
  case PathList("META-INF", xs @ _*) => MergeStrategy.discard
  case "reference.conf"              => MergeStrategy.concat
  case x                             => MergeStrategy.first
}

sbt assembly
# → target/scala-2.12/myapp-assembly-1.0.jar

spark-submit \
  --class com.company.MainApp \
  --master yarn \
  target/scala-2.12/myapp-assembly-1.0.jar

Maven + Shade Plugin:

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-shade-plugin</artifactId>
  <version>3.5.1</version>
  <executions>
    <execution>
      <phase>package</phase>
      <goals><goal>shade</goal></goals>
      <configuration>
        <filters>
          <filter>
            <artifact>*:*</artifact>
            <excludes>
              <exclude>META-INF/*.SF</exclude>
              <exclude>META-INF/*.DSA</exclude>
            </excludes>
          </filter>
        </filters>
      </configuration>
    </execution>
  </executions>
</plugin>

mvn package -DskipTests
spark-submit --class com.company.Main target/myapp-1.0-shaded.jar

Тонкий JAR + --jars / --packages¶

Если зависимости уже есть на кластере или их нужно тянуть из Maven:

# Локальные JAR-файлы
spark-submit \
  --class com.company.Main \
  --jars lib/custom-connector.jar,lib/extra-lib.jar \
  app-thin.jar

# Maven Central через --packages
spark-submit \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0,\
io.delta:delta-spark_2.12:3.1.0 \
  --class com.company.Main \
  app.jar

--packages при первом запуске скачивает JAR в локальный Maven-кэш (~/.ivy2/). При повторных запусках использует кэш. В air-gapped кластерах нужен --repositories с внутренним Nexus/Artifactory.

Dependency Hell и relocation¶

Проблема: ваш JAR зависит от guava:20, а сам Spark использует guava:16. Конфликт классов → ClassCastException или NoSuchMethodError.

Решение - relocation (shadowing): переименовать пакет зависимости внутри fat JAR:

<!-- Maven Shade Plugin: relocate guava -->
<relocations>
  <relocation>
    <pattern>com.google.common</pattern>
    <shadedPattern>com.company.shaded.google.common</shadedPattern>
  </relocation>
</relocations>

// sbt-assembly: relocate guava
assemblyShadeRules := Seq(
  ShadeRule.rename("com.google.common.**" ->
    "com.company.shaded.google.@1").inAll
)

После relocation ваш код использует com.company.shaded.google.common.*, а Spark - свой com.google.common.*. Конфликтов нет.

--packages vs --jars vs --files: что когда¶

Чтение --files на Executor:

# Spark раскладывает --files в текущий рабочий каталог Executor
# Читать нужно по имени файла, не по пути
import yaml

# spark-submit --files config/production.yaml
with open("production.yaml") as f:  # только имя файла!
    config = yaml.safe_load(f)

Управление ресурсами¶

Формула расчёта памяти Executor¶

container_memory = executor-memory + memoryOverhead

memoryOverhead = max(executor-memory × 0.1, 384MB)

Пример: --executor-memory 8g → container = 8192 + 819 ≈ 9011 MB. YARN выделит контейнер ≥ 9011 MB.

Параметр	Default	Рекомендация
`spark.executor.memory`	1g	4–16g в зависимости от данных
`spark.executor.memoryOverhead`	max(10%, 384m)	Увеличить при OOM в Pandas UDF
`spark.driver.memory`	1g	4–8g если `collect()` или большой broadcast
`spark.driver.maxResultSize`	1g	Увеличить или `0` для отключения лимита

Executor cores и параллелизм¶

effective_parallelism = num-executors × executor-cores

Правило: 4–5 cores на Executor - оптимум. Меньше → меньше параллелизма. Больше → конкуренция за ресурсы JVM, GC-давление.

# Плохо: 1 executor × 40 cores
--num-executors 1 --executor-cores 40

# Хорошо: 10 executors × 4 cores = 40 параллельных задач
--num-executors 10 --executor-cores 4 --executor-memory 8g

Dynamic Allocation¶

Если не задавать --num-executors, Spark автоматически масштабирует количество Executor-ов:

spark-submit \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.minExecutors=2 \
  --conf spark.dynamicAllocation.maxExecutors=50 \
  --conf spark.dynamicAllocation.initialExecutors=5 \
  main.py

Dynamic Allocation требует spark.shuffle.service.enabled=true на NodeManager-ах (или External Shuffle Service на K8s).

Запуск на разных кластер-менеджерах¶

YARN¶

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --queue data-team \
  --conf spark.yarn.tags=etl,daily \
  --conf spark.yarn.maxAppAttempts=3 \
  main.py

Проверка статуса:

yarn application -status application_1234567890_0001

Kubernetes¶

spark-submit \
  --master k8s://https://k8s-api:6443 \
  --deploy-mode cluster \
  --conf spark.kubernetes.container.image=myrepo/spark:3.5.1 \
  --conf spark.kubernetes.namespace=spark-jobs \
  --conf spark.kubernetes.driver.request.cores=1 \
  --conf spark.kubernetes.executor.request.cores=2 \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
  local:///opt/spark/work-dir/main.py

Обратите внимание: local:/// - путь внутри Docker-образа, а не на вашей машине.

Standalone¶

spark-submit \
  --master spark://master-host:7077 \
  --deploy-mode cluster \
  --supervise \
  main.py

--supervise - автоматический перезапуск Driver при падении (только Standalone).

Порядок загрузки classpath¶

Spark разрешает конфликты версий по следующей иерархии:

Если ваш JAR конфликтует со Spark-внутренним - добавьте в начало SPARK_CLASSPATH или используйте relocation.

Логирование и диагностика¶

Где искать логи¶

YARN: извлечение логов¶

# Все логи приложения (Driver + все Executor)
yarn logs -applicationId application_1234567890_0001

# Только Driver (ApplicationMaster)
yarn logs -applicationId application_1234567890_0001 \
  -containerId container_1234567890_0001_01_000001

# Только stdout конкретного контейнера
yarn logs -applicationId application_1234567890_0001 \
  -containerId container_1234567890_0001_01_000002 \
  -log_files stdout

# Следить в реальном времени (cluster mode)
yarn logs -applicationId application_1234567890_0001 -follow

Найти Spark UI в cluster mode¶

# 1. Через yarn application
yarn application -status application_1234567890_0001 | grep "Tracking URL"

# 2. Через REST API YARN RM
curl http://yarn-rm:8088/ws/v1/cluster/apps/application_1234567890_0001 \
  | python3 -m json.tool | grep trackingUrl

# 3. Spark History Server (после завершения)
# http://history-server:18080/history/application_1234567890_0001/

Типичные ошибки запуска¶

ClassNotFoundException / NoClassDefFoundError

Exception in thread "main" java.lang.ClassNotFoundException: com.example.Main

Причина: неправильный --class, JAR не попал в classpath, или fat JAR не собран.

# Проверить что в JAR есть нужный класс
jar tf app.jar | grep "com/example/Main"

# Проверить --class совпадает с полным именем
spark-submit --class com.example.Main ...  # не Main, а com.example.Main

ModuleNotFoundError (Python)

ModuleNotFoundError: No module named 'pandas'

Причина: --py-files добавляет файлы в sys.path, но не устанавливает пакеты. Pandas не поставить через --py-files.

# Решение 1: conda-pack или venv + --archives
# Решение 2: предустановить на всех воркерах
# Решение 3: для PySpark 3.1+
spark-submit --conf spark.pyspark.pip.install="pandas==2.1.0,pyarrow==14.0" main.py

Версии Python Driver vs Executor

Exception: Python worker failed to connect back

Причина: Driver и Executor используют разные версии Python.

spark-submit \
  --conf spark.pyspark.python=/usr/bin/python3.10 \
  --conf spark.pyspark.driver.python=/usr/bin/python3.10 \
  main.py

Container killed by YARN for exceeding memory limits

Container [pid=12345,containerID=container_xxx] is running
beyond virtual memory limits. Current usage: 8.5 GB of 8 GB

Причина: JVM использует больше памяти чем executor-memory. Нужно увеличить memoryOverhead.

spark-submit \
  --executor-memory 8g \
  --conf spark.executor.memoryOverhead=2g \
  main.py

Практика: один скрипт - разные окружения¶

# main.py
import argparse
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count


def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument("--input", required=True)
    p.add_argument("--output", required=True)
    p.add_argument("--date", required=True)
    return p.parse_args()


def run(spark: SparkSession, args) -> None:
    df = spark.read.parquet(args.input)
    result = (
        df.filter(col("date") == args.date)
          .groupBy("user_id")
          .agg(count("*").alias("events"))
    )
    result.write.mode("overwrite").parquet(args.output)


if __name__ == "__main__":
    args = parse_args()
    spark = SparkSession.builder.appName("daily-events-agg").getOrCreate()
    run(spark, args)
    spark.stop()

# Локальная разработка
spark-submit \
  --master local[*] \
  --conf spark.sql.shuffle.partitions=4 \
  main.py \
  --input data/sample/ \
  --output /tmp/output/ \
  --date 2024-01-15

# Dev YARN (cluster mode)
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 4g \
  --executor-cores 2 \
  --num-executors 3 \
  --archives spark-env.tar.gz#spark-env \
  --conf spark.executorEnv.PYSPARK_PYTHON=./spark-env/bin/python \
  main.py \
  --input hdfs:///data/events/ \
  --output hdfs:///output/daily-agg/ \
  --date 2024-01-15

# Production YARN + Iceberg
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name "daily-events-agg-2024-01-15" \
  --queue data-engineering \
  --executor-memory 16g \
  --executor-cores 4 \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.maxExecutors=30 \
  --conf spark.sql.shuffle.partitions=400 \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0 \
  --archives spark-env.tar.gz#spark-env \
  --conf spark.executorEnv.PYSPARK_PYTHON=./spark-env/bin/python \
  --files config/production.yaml \
  main.py \
  --input s3a://datalake/events/ \
  --output s3a://datalake/daily-agg/ \
  --date 2024-01-15

Production Best Practices¶

Версионирование артефактов

# Плохо: нет версии, нельзя откатить
spark-submit main.py

# Хорошо: версия в имени + CI/CD тег
spark-submit \
  --name "etl-pipeline-v1.4.2-$(date +%Y%m%d)" \
  hdfs:///apps/etl-pipeline-1.4.2.zip

Reproducible builds

# Зафиксировать версии зависимостей
conda env export > environment.lock.yml
pip freeze > requirements.lock.txt

# Сборка воспроизводима: используйте lock-файлы, не диапазоны версий
conda env create -f environment.lock.yml
conda pack -o spark-env-1.4.2.tar.gz

Минимизация зависимостей

# Проверить что реально используется
pipdeptree --warn silence | grep -v "^  "

# Не включать dev-зависимости в spark-env
pip install --no-dev-deps -r requirements.txt

Шаблон команды запуска

#!/bin/bash
set -euo pipefail

APP_VERSION="${APP_VERSION:-1.0.0}"
ENV="${ENV:-dev}"
DATE="${1:-$(date +%Y-%m-%d -d yesterday)}"

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name "etl-${ENV}-${APP_VERSION}-${DATE}" \
  --queue "data-${ENV}" \
  --executor-memory "${EXECUTOR_MEMORY:-8g}" \
  --executor-cores "${EXECUTOR_CORES:-4}" \
  --conf "spark.dynamicAllocation.enabled=true" \
  --conf "spark.dynamicAllocation.maxExecutors=${MAX_EXECUTORS:-20}" \
  --conf "spark.sql.shuffle.partitions=${SHUFFLE_PARTITIONS:-200}" \
  --conf "spark.eventLog.enabled=true" \
  --conf "spark.eventLog.dir=hdfs:///spark-history/" \
  --archives "hdfs:///envs/spark-env-${APP_VERSION}.tar.gz#spark-env" \
  --conf "spark.executorEnv.PYSPARK_PYTHON=./spark-env/bin/python" \
  "hdfs:///apps/etl-${APP_VERSION}/main.py" \
  --date "${DATE}" \
  --env "${ENV}"

Event Log - обязательно включать

--conf spark.eventLog.enabled=true
--conf spark.eventLog.dir=hdfs:///spark-history/

Без Event Log History Server не покажет завершённое приложение - невозможно диагностировать проблемы после факта.

Итог¶

spark-submit - точка сборки всего: код, зависимости, конфигурация, ресурсы. Ключевые решения при каждом запуске:

deploy-mode: client для отладки, cluster для production
упаковка Python: --py-files для модулей, --archives для полного окружения
JVM зависимости: fat JAR для изоляции, --packages для connectors
ресурсы: 4–5 cores на Executor, overhead ≥ 10%, Dynamic Allocation для production
логи: Event Log всегда включать, YARN logs для post-mortem диагностики

Типичные ошибки - ClassNotFoundException, ModuleNotFoundError, OOM из-за memoryOverhead - решаются пониманием того, как Spark доставляет артефакты через Staging Area и как устроен classpath.