Spark cluster локально: master + workers

Apache Spark — это распределённый compute-engine. Локально на ноутбуке его обычно запускают в standalone mode: один master-процесс координирует, два-три worker-процесса делают работу. Это упрощённая модель кластера: на проде вместо standalone обычно YARN или Kubernetes, но логика та же — master раздаёт таски, worker’ы выполняют.

В этом уроке поднимем такой кластер в compose: один master, два worker’а, и научимся сабмитить Python-job из примонтированного volume.

Что делает планировщик — preemptive vs cooperative, time slice

Архитектура standalone-кластера

В standalone Spark:

Master — единая точка координации. Слушает 7077 для приёма spark-submit, 8080 для Web UI.
Worker — executor host. Регистрируется у master’а, получает таски, отчитывается о ресурсах (cores, memory).
Driver — процесс твоего Spark-приложения. Создаёт SparkContext, разбивает работу на stages/tasks, отправляет worker’ам. Может крутиться внутри master-контейнера (--deploy-mode cluster) или на клиенте (--deploy-mode client — дефолт).

Spark standalone: master + 2 workers + driver

spark-master:7077 RPC, :8080 UI

spark-worker-1

spark-worker-2

driver (client)

tasks

executor

Compose-файл

services:
  spark-master:
    image: bitnami/spark:3.5
    container_name: spark-master
    environment:
      SPARK_MODE: master
      SPARK_RPC_AUTHENTICATION_ENABLED: 'no'
      SPARK_RPC_ENCRYPTION_ENABLED: 'no'
      SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED: 'no'
      SPARK_SSL_ENABLED: 'no'
    ports:
      - "7077:7077"
      - "8080:8080"
    volumes:
      - ./app:/opt/app
      - ./data:/opt/data
    networks:
      - spark-net

  spark-worker-1:
    image: bitnami/spark:3.5
    container_name: spark-worker-1
    environment:
      SPARK_MODE: worker
      SPARK_MASTER_URL: spark://spark-master:7077
      SPARK_WORKER_MEMORY: 2G
      SPARK_WORKER_CORES: 2
      SPARK_RPC_AUTHENTICATION_ENABLED: 'no'
      SPARK_SSL_ENABLED: 'no'
    depends_on:
      - spark-master
    volumes:
      - ./app:/opt/app
      - ./data:/opt/data
    networks:
      - spark-net

  spark-worker-2:
    image: bitnami/spark:3.5
    container_name: spark-worker-2
    environment:
      SPARK_MODE: worker
      SPARK_MASTER_URL: spark://spark-master:7077
      SPARK_WORKER_MEMORY: 2G
      SPARK_WORKER_CORES: 2
      SPARK_RPC_AUTHENTICATION_ENABLED: 'no'
      SPARK_SSL_ENABLED: 'no'
    depends_on:
      - spark-master
    volumes:
      - ./app:/opt/app
      - ./data:/opt/data
    networks:
      - spark-net

networks:
  spark-net:

Что важно:

Один образ для всех — bitnami/spark:3.5. SPARK_MODE: master|worker переключает поведение.
SPARK_MASTER_URL: spark://spark-master:7077 — worker’ы находят master по DNS-имени compose-сервиса.
SPARK_WORKER_MEMORY и SPARK_WORKER_CORES — ресурсы каждого worker’а. На ноутбуке 16GB / 8 cores: 2 worker’а по 2GB / 2 cores — безопасно.
./app:/opt/app — bind mount для твоих Python-файлов. Чтобы spark-submit видел их.
./data:/opt/data — bind mount для данных (CSV, parquet). Не запихиваем большие dataset’ы в образ.

Запуск кластера

mkdir -p app data

docker compose up -d

# Проверка
docker compose ps
# spark-master      Up      0.0.0.0:7077->7077/tcp, 0.0.0.0:8080->8080/tcp
# spark-worker-1    Up
# spark-worker-2    Up

# Открой Web UI
open http://localhost:8080
# Увидишь:
# URL: spark://spark-master:7077
# Workers: 2 (alive)
# Cores: 4 total, 0 used
# Memory: 4 GB total

В Web UI Workers должны показывать состояние ALIVE и общий resource pool. Если worker’ов 0 — проверь логи (docker compose logs spark-worker-1).

Первый job: word count на Python

Создай файл ./app/wordcount.py:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("WordCount")
    .getOrCreate()
)

# Читаем текстовый файл (надо положить ./data/sample.txt)
df = spark.read.text("/opt/data/sample.txt")

# Разбиваем на слова, считаем
from pyspark.sql.functions import explode, split, count

word_counts = (
    df
    .select(explode(split(df.value, " ")).alias("word"))
    .groupBy("word")
    .agg(count("*").alias("cnt"))
    .orderBy("cnt", ascending=False)
)

word_counts.show(20, truncate=False)
spark.stop()

И семпл-файл ./data/sample.txt:

docker docker compose docker
spark spark master spark worker
junior data engineer junior
docker for junior data engineer

Submit job через spark-submit

spark-submit запускается внутри master-контейнера (или из отдельного клиента — но проще из master’а, где Spark уже установлен):

docker compose exec spark-master spark-submit \
  --master spark://spark-master:7077 \
  --deploy-mode client \
  --total-executor-cores 4 \
  --executor-memory 1G \
  /opt/app/wordcount.py

Разбор флагов:

--master spark://spark-master:7077 — куда сабмитить (наш master по DNS).
--deploy-mode client — driver работает в том же процессе, что и spark-submit. Логи видны сразу. Альтернатива cluster — driver запускается на одном из worker’ов.
--total-executor-cores 4 — суммарно executor’ы получат до 4 cores (мы дали 2+2 = 4 worker cores).
--executor-memory 1G — память на каждый executor. Меньше, чем worker memory (2G), потому что worker может запустить >1 executor.
/opt/app/wordcount.py — путь внутри контейнера (наш volume mount).

После запуска ты увидишь в выводе:

+----------+---+
|word      |cnt|
+----------+---+
|docker    |4  |
|junior    |3  |
|spark     |3  |
|data      |2  |
|engineer  |2  |
|...       |...|
+----------+---+

А в Web UI (http://localhost:8080) появится Application: WordCount со статусом RUNNING / FINISHED, длительностью и распределением task’ов по executor’ам.

Standalone vs Kubernetes — когда переходить

Standalone (compose) vs Kubernetes Spark Operator

Standalone в composedev, тест, малые job'ы

growth

Spark on k8sprod, ETL pipelines

Local dev

Managed (EMR/Dataproc)

Standalone-кластер в compose даёт тебе:

Полную копию production API (DataFrame, SparkSession, spark-submit — те же).
Быструю обратную связь: правишь .py файл — снова делаешь spark-submit.
Никакой настройки k8s, namespace’ов, RBAC, image pull secrets.

Когда понадобится k8s:

Когда job’ам нужен autoscaling (10 executor’ов для одной job’ы, 100 для другой).
Когда несколько команд делят кластер (multi-tenancy через namespaces).
Когда нужен rolling-upgrade Spark-версии без downtime.

Для большинства junior-задач compose-кластера хватит на месяцы вперёд.

WARNING

ВНИМАНИЕ: bitnami/spark Docker-образ — это для dev. В prod не используй его как есть: нет встроенного S3-клиента, нет JDBC-драйверов, нет Iceberg/Delta connector. Собирай свой образ FROM bitnami/spark:3.5 c нужными JAR-ами через —packages или COPY.

Подключение Postgres / MinIO к Spark

Чаще всего junior’у нужен Spark с источником данных в Postgres или S3-совместимом сторадже (MinIO). Для этого нужен JDBC-драйвер (для Postgres) или Hadoop S3A connector (для MinIO).

Самый простой способ — через --packages:

docker compose exec spark-master spark-submit \
  --master spark://spark-master:7077 \
  --packages org.postgresql:postgresql:42.7.4,org.apache.hadoop:hadoop-aws:3.3.4 \
  --conf spark.hadoop.fs.s3a.endpoint=http://minio:9000 \
  --conf spark.hadoop.fs.s3a.path.style.access=true \
  /opt/app/etl_job.py

При первом запуске Spark скачает JAR’ы из Maven Central — это медленно. Чтобы кешировать, добавь volume:

spark-master:
  volumes:
    - spark-ivy:/opt/bitnami/spark/.ivy2

И аналогично для worker’ов (они тоже могут скачивать). Названный volume spark-ivy сохранит загруженные JAR’ы между перезапусками.

Trade-offs локального стенда

Что хорошо:

Один YAML — весь кластер.
Полная Spark API параллельность (worker’ы реально работают параллельно).
Можно проверить логику разбиения на партиции, broadcast joins, агрегации.

Что плохо:

Все worker’ы делят CPU и память с master’ом и хост-машиной. На ноутбуке с 16GB RAM реально можно нагрузить только ~8GB Spark’а.
Нет реального distributed shuffle через сеть (всё в одной docker-сети, fast loopback).
Нет HDFS / S3 — придётся mount’ить bind volumes или поднимать MinIO.

Trade-off очевиден: compose-стенд это dev-environment, не prod. Job’ы, которые работают на compose, в prod почти всегда работают тоже — но не наоборот (prod scale может вскрыть проблемы, не видимые локально).

Попробуй сам

# 1. Запусти кластер
docker compose up -d

# 2. UI
open http://localhost:8080
# 2 workers ALIVE, 4 cores, 4GB memory

# 3. Создай ./app/wordcount.py и ./data/sample.txt (см. выше)

# 4. Submit
docker compose exec spark-master spark-submit \
  --master spark://spark-master:7077 \
  --total-executor-cores 4 \
  --executor-memory 1G \
  /opt/app/wordcount.py

# 5. В UI -> Completed Applications -- увидишь свой job

# 6. Попробуй scale-up workers
docker compose up -d --scale spark-worker-1=2
# Но: имена контейнеров конфликтуют. Удалить container_name и попробовать снова.

# 7. Cleanup
docker compose down

Проверка знанийKnowledge check

Ты запустил spark-submit и job упал с ошибкой "Initial job has not accepted any resources; check your cluster UI". В UI workers ALIVE, но cores 0/4 used и Memory Free: 4GB. Что с ресурсами?

ОтветAnswer

Ты запросил больше ресурсов, чем кластер может выделить ОДНОМУ executor'у. Например: --executor-memory 3G, но каждый worker даёт 2GB. Spark не может запустить ни одного executor'а -- ни один worker не подходит. Решения: (1) уменьшить --executor-memory до значения, помещающегося в worker (например, 1G); (2) увеличить SPARK_WORKER_MEMORY в compose; (3) проверить --total-executor-cores -- если он больше суммы worker cores, тоже не получит ресурсы. Spark показывает this Pending в Applications -> Running -- внутри увидишь "Application is added to the scheduler and is not yet activated".