fix(spark): исправлено подключение ноутбуков к кластеру

- Зачем: модули 2–8 в новом процессе выбирали local[*] вместо Spark-кластера. - Что: master перенесён в общий spark-defaults.conf, smoke требует точный адрес кластера, документация и планы синхронизированы. - Проверка: фактическая инициализация всех восьми ноутбуков, повторный getOrCreate, распределённые вычисления, Iceberg/MinIO write-read-cleanup и docker compose config прошли.
2026-09-15 08:18:33 -04:00
1 parent a085b2fc55
commit 91a5c85469
@@ -56,6 +56,10 @@ graph TB
 | `Trino` | Compute — ad hoc SQL, чтение тех же таблиц |
 | `JupyterLab` | Точка входа — практические ноутбуки курса |

+Ноутбуки создают `SparkSession` с именем приложения, а подключение к
+`spark://spark-master:7077` и общие настройки Iceberg/MinIO получают из
+`spark/spark-defaults.conf`.
+
 ## С чего начать

 1. Открой [START_HERE.md](./START_HERE.md) — запуск стенда, диагностика, первый ноутбук.
@@ -108,6 +108,8 @@ notebooks/01_environment_and_smoke_test.ipynb
 - `./src` смонтирован read-only как `/opt/src`;
 - `./data` смонтирован read-only как `/opt/data`;
 - `PYTHONPATH=/opt/src`, поэтому helper-скрипты из `src/` доступны для импорта в ноутбуках;
+- новое ядро Jupyter читает `spark/spark-defaults.conf`: из него `SparkSession` получает адрес кластера `spark://spark-master:7077` и настройки Iceberg/MinIO;
+- повторный запуск ячейки с `getOrCreate()` в том же ядре возвращает существующую сессию; чтобы заново применить стартовые настройки, перезапусти ядро;
 - первый ноутбук курса: `01_environment_and_smoke_test.ipynb`;
 - локальный учебный data bundle для Модуля 3 должен быть доступен внутри Jupyter по пути `/opt/data/nyc_taxi`.

@@ -54,6 +54,8 @@ docker compose down -v
 - Master UI: `http://localhost:8080`
 - Worker UI: `http://localhost:8081` и `http://localhost:8082`
 - Spark master endpoint внутри сети compose: `spark://spark-master:7077`
+- `spark/spark-defaults.conf` задаёт этот endpoint как `spark.master`; новое ядро Jupyter применяет настройку при создании первой `SparkSession`.
+- `SparkSession.builder.getOrCreate()` в уже работающем ядре возвращает существующую сессию. Для повторного чтения стартовых настроек нужно перезапустить ядро.

 Пример запуска smoke-скрипта:

@@ -141,7 +143,7 @@ docker compose run --rm iceberg-catalog-init
 | Файл | Что задаёт |
 | --- | --- |
 | `docker-compose.yml` | состав сервисов, порты, volumes, init-контейнеры |
-| `spark/spark-defaults.conf` | Spark catalog `lakehouse`, `s3a`, Iceberg extensions |
+| `spark/spark-defaults.conf` | Spark master, catalog `lakehouse`, `s3a`, Iceberg extensions |
 | `trino/catalog/lakehouse.properties` | каталог Trino `lakehouse`, JDBC и S3-доступ |
 | `spark/Dockerfile` | образ Spark с Iceberg, S3A и Python-зависимостями |
 | `jupyter/Dockerfile` | образ JupyterLab на базе Spark |
@@ -25,8 +25,8 @@
      "metadata": {},
      "outputs": [],
      "source": [
-        "# Мы не указываем .master() явно, так как он и другие настройки Iceberg\n",
-        "# подтягиваются из конфигурации spark-defaults.conf при старте сессии.\n",
+        "# Адрес Spark-кластера и настройки Iceberg берутся из spark-defaults.conf.\n",
+        "# Поэтому в ноутбуке достаточно задать имя приложения.\n",
        "from pyspark.sql import SparkSession\n",
        "\n",
        "spark = SparkSession.builder \\\n",
@@ -22,15 +22,13 @@

 ### 1. Проверить и исправить создание Spark-сессий

-**Статус:** запланировано; выполняется первым.
+**Статус:** выполнено в задаче #1.

 **Задача:** [#1 — подключение Spark-сессий в новом ядре](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/1).

-В `src/spark/cluster_smoke.py` адрес master задаётся явно. В модуле 2 комментарий говорит, что master берётся из `spark-defaults.conf`, однако в этом файле нет `spark.master`. В модулях 2–8 сессия создаётся без явного `.master(...)`. Это подтверждённое расхождение исходников; фактический режим работы этих ноутбуков в новом ядре ещё не проверен.
+Проверка в отдельных процессах показала: модуль 1 подключался к кластеру, а модули 2–8 без `spark.master` запускали Spark в `local[*]`. Теперь общий `spark-defaults.conf` задаёт `spark://spark-master:7077`, а helper модуля 1 и остальные ноутбуки используют этот единый источник.

-Нужно проверить создание сессии каждого ноутбука в отдельном новом ядре, зафиксировать фактический master и привести код и пояснения к воспроизводимому способу подключения. Повторное прохождение сохраняет учебные данные из предыдущих модулей: независимость от состояния ядра не означает независимость от этих данных.
-
-**Результат:** каждый ноутбук подключается к ожидаемому Spark-кластеру после перезапуска ядра; документация правильно объясняет источник настроек. Проверки и необходимые исправления описываются в задаче.
+**Результат:** каждый ноутбук подключается к ожидаемому Spark-кластеру после перезапуска ядра. Повторный `getOrCreate()` в том же ядре возвращает существующую сессию; это поведение и источник настроек объяснены в onboarding и техническом reference. Подробные проверки сохранены в задаче #1.

 ### 2. Добавить практику «Как настроен Spark в нашем стенде»

@@ -96,7 +96,7 @@ volumes:

 ### Секция 3: Первое чтение в Spark
 - **[md]** Разница `spark.read.parquet("s3a://...")` (raw-файлы) vs `spark.table(...)` (управляемая Iceberg-таблица). Мы пока на этапе raw, Iceberg-таблица появится в Модуле 4.
- **[code]** Создание SparkSession (паттерн из Module 2 — без `.master()`, `setLogLevel("ERROR")`).
+- **[code]** Создание SparkSession (паттерн из Module 2 — адрес кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`).
 - **[code]** Чтение одного файла: `spark.read.parquet("s3a://lakehouse/raw/nyc_taxi/yellow_tripdata_2024-01.parquet")`.
 - **[code]** Чтение всех файлов wildcard: `yellow_tripdata_*.parquet` (без привязки к конкретному году в wildcard, чтобы код работал с любыми скачанными файлами).

@@ -150,7 +150,7 @@ volumes:
 - `.gitignore` содержит `/data/` для исключения датасета из git.
 - `notebooks/03_raw_ingest_and_first_read.ipynb` можно выполнить сверху вниз в поднятом Jupyter-окружении при наличии скачанного data bundle.
 - Ноутбук следует методике курса: `объяснение -> демонстрация -> самостоятельное повторение -> checkpoint`.
- Ноутбук использует паттерны из Модулей 1-2: `boto3` для MinIO, SparkSession без `.master()`, `setLogLevel("ERROR")`.
+- Ноутбук использует паттерны из Модулей 1-2: `boto3` для MinIO, SparkSession с адресом кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`.
 - Raw-данные остаются в MinIO после выполнения ноутбука (cleanup отсутствует намеренно).
 - При отсутствии data bundle — понятный assert с отсылкой к `START_HERE.md`.
 - `docs/stack_reference.md` обновлён: секция JupyterLab содержит `./data` mount.
@@ -94,7 +94,7 @@ Bronze нужен Модулю 5. Spark-сессия останавливает
 | История изменений | Нет (или WAL на уровне СУБД) | Snapshots и manifests (встроены в формат) |

 ### Секция 1: Spark-сессия и проверка raw
- **[code]** SparkSession (паттерн Модулей 2-3: без `.master()`, `setLogLevel("ERROR")`).
+- **[code]** SparkSession (паттерн Модулей 2-3: адрес кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`).
 - **[code]** Быстрая проверка: `spark.read.parquet("s3a://lakehouse/raw/nyc_taxi/yellow_tripdata_*.parquet").count()` + `.printSchema()`. Assert на ненулевой count с отсылкой к Модулю 3.
 - **[md]** Сейчас читаем как набор файлов — нет истории, каталогизации, контроля схемы. Дальше создадим управляемую таблицу.

@@ -203,7 +203,7 @@ Bronze нужен Модулю 5. Spark-сессия останавливает

 - `notebooks/04_bronze_with_iceberg.ipynb` можно выполнить сверху вниз в поднятом Jupyter-окружении после прохождения Модуля 3 (raw-данные в MinIO).
 - Ноутбук следует методике курса: `объяснение → демонстрация → самостоятельное повторение → checkpoint`.
- Ноутбук использует паттерны из Модулей 1-3: SparkSession без `.master()`, `setLogLevel("ERROR")`, `boto3` для MinIO.
+- Ноутбук использует паттерны из Модулей 1-3: SparkSession с адресом кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`, `boto3` для MinIO.
 - Ноутбук идемпотентен: повторное выполнение (`CREATE OR REPLACE`) не дублирует данные и не роняет ячейки.
 - CTAS использует явный список колонок, а не `SELECT *`.
 - Bronze-таблица `lakehouse.bronze.nyc_taxi_yellow` остаётся после выполнения ноутбука для использования в Модуле 5.
@@ -118,7 +118,7 @@ Silver нужен Модулям 6 и 8. Spark-сессия останавлив
 - **[md]** Ключевая идея: silver = bronze + явные правила трансформации + проверки качества.

 ### Секция 1: Spark-сессия и проверка bronze
- **[code]** SparkSession (паттерн Модулей 2-4: без `.master()`, `setLogLevel("ERROR")`). Константы, вспомогательные функции (`format_bytes`, `list_objects` — переиспользование паттерна из Модуля 4).
+- **[code]** SparkSession (паттерн Модулей 2-4: адрес кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`). Константы, вспомогательные функции (`format_bytes`, `list_objects` — переиспользование паттерна из Модуля 4).
 - **[code]** Assert: `spark.table(BRONZE_TABLE).count() > 0` с отсылкой к Модулю 4. Assert: `spark.table(BRONZE_LOOKUP_TABLE)` существует — при отсутствии жёсткая ошибка с отсылкой к Секции 8 Модуля 4.
 - **[md]** Обе bronze-таблицы на месте. Теперь строим silver.

@@ -260,7 +260,7 @@ speed_mph                   DOUBLE          (E4: новая, trip_distance / (tr

 - `notebooks/05_silver_layer.ipynb` можно выполнить сверху вниз в поднятом Jupyter-окружении после прохождения Модуля 4 (bronze-таблицы в каталоге).
 - Ноутбук следует методике курса: `объяснение -> демонстрация -> самостоятельное повторение -> checkpoint`.
- Ноутбук использует паттерны из Модулей 1-4: SparkSession без `.master()`, `setLogLevel("ERROR")`.
+- Ноутбук использует паттерны из Модулей 1-4: SparkSession с адресом кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`.
 - Ноутбук идемпотентен: повторное выполнение (`CREATE OR REPLACE`) не дублирует данные и не роняет ячейки.
 - CTAS использует явный список колонок, а не `SELECT *`.
 - Правила трансформации зафиксированы в markdown-таблице перед кодом, а не только в коде.
@@ -287,7 +287,7 @@ RUN pip3 install --no-cache-dir \
 - `notebooks/06_spark_and_trino_on_same_table.ipynb` выполняется сверху вниз после прохождения Модуля 5 без внешних зависимостей (DBeaver не обязателен).
 - Ноутбук содержит живой цикл write → read: Spark создаёт таблицу, Trino читает.
 - Методика курса: объяснение -> демонстрация -> самостоятельное повторение -> checkpoint.
- Паттерны Модулей 1-5: SparkSession без `.master()`, `setLogLevel("ERROR")`.
+- Паттерны Модулей 1-5: SparkSession с адресом кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`.
 - Trino-запросы — в исполняемых code cells через Python `trino` клиент. DBeaver-подсказки — в markdown.
 - При отсутствии silver — assert с отсылкой к Модулю 5. Схема silver: не менее 24 колонок (допускается расширенная схема после Модуля 5).
 - Текст на русском с параллелями к PostgreSQL/Greenplum.
@@ -429,7 +429,7 @@ print(f"Демо-таблица пересоздана: {spark.table(DEMO_TABLE)
 - `notebooks/07_safe_table_changes.ipynb` выполняется сверху вниз после прохождения Модуля 6 без внешних зависимостей (DBeaver не обязателен).
 - Ноутбук содержит: демонстрацию schema evolution (`ADD COLUMN` на silver), time travel (`VERSION AS OF`), recovery (`rollback_to_snapshot`), демонстрацию влияния `RENAME COLUMN` на downstream.
 - Методика курса: объяснение -> демонстрация -> самостоятельное повторение -> checkpoint.
- Паттерны Модулей 1-6: SparkSession без `.master()`, `setLogLevel("ERROR")`, `trino_query()`.
+- Паттерны Модулей 1-6: SparkSession с адресом кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`, `trino_query()`.
 - Schema evolution на silver: только `ADD COLUMN` (безопасная операция). Опасные операции — на демо-таблице.
 - При отсутствии silver — assert с отсылкой к Модулю 5.
 - Текст на русском с параллелями к PostgreSQL/Greenplum.
@@ -488,7 +488,7 @@ except Exception as e:
 - `notebooks/08_maintenance_and_final_lab.ipynb` выполняется сверху вниз после прохождения Модуля 6 без внешних зависимостей (DBeaver не обязателен, Модуль 7 рекомендуется, но не обязателен).
 - Ноутбук содержит: демонстрацию проблемы мелких файлов, compaction (`rewrite_data_files`), expire_snapshots, финальную сквозную практику.
 - Методика курса: объяснение → демонстрация → самостоятельное повторение → checkpoint.
- Паттерны Модулей 1–7: SparkSession без `.master()`, `setLogLevel("ERROR")`, `trino_query()`.
+- Паттерны Модулей 1–7: SparkSession с адресом кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`, `trino_query()`.
 - Демо-таблица `taxi_maintenance_demo` создаётся и удаляется внутри ноутбука.
 - Финальная практика в отдельном namespace `lakehouse.final` с полным cleanup в конце.
 - При отсутствии silver — assert с отсылкой к Модулю 5. При отсутствии raw-данных в MinIO — assert с отсылкой к Модулю 3.
@@ -1,5 +1,8 @@
 # spark/spark-defaults.conf

+# Все SparkSession из Jupyter подключаются к кластеру по умолчанию
+spark.master                                     spark://spark-master:7077
+
 # Включаем Iceberg extensions
 spark.sql.extensions                             org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

@@ -2,30 +2,30 @@ from __future__ import annotations

 from pyspark.sql import SparkSession

-DEFAULT_MASTER_URL = "spark://spark-master:7077"
+EXPECTED_MASTER_URL = "spark://spark-master:7077"
 DEFAULT_APP_NAME = "cluster-smoke"
 DEFAULT_ROW_COUNT = 1_000_000


 def create_spark_session(
    app_name: str = DEFAULT_APP_NAME,
-    master_url: str = DEFAULT_MASTER_URL,
 ) -> SparkSession:
    """Создать SparkSession для проверок окружения в Модуле 1.

    Функция специально остаётся минимальной и опирается на смонтированный
-    в контейнер `spark-defaults.conf` для остальной Spark/Iceberg-конфигурации.
+    в контейнер `spark-defaults.conf` для Spark/Iceberg-конфигурации.
    """
-    return SparkSession.builder.appName(app_name).master(master_url).getOrCreate()
+    return SparkSession.builder.appName(app_name).getOrCreate()


 def run_cluster_smoke(spark: SparkSession, row_count: int = DEFAULT_ROW_COUNT) -> dict:
    spark_context = spark.sparkContext
    master_url = spark_context.master

-    if master_url.startswith("local"):
+    if master_url != EXPECTED_MASTER_URL:
        raise RuntimeError(
-            "Smoke test must run against the Spark cluster, not local mode."
+            "Smoke test must run against "
+            f"{EXPECTED_MASTER_URL}, got {master_url}."
        )

    # Берём больше партиций, чем даёт defaultParallelism, чтобы задача точно