Merge pull request 'feat(course): практика настройки SparkSession в модуле 3' (#4) from 2-spark-session-practice into main

This commit was merged in pull request #4.
This commit is contained in:
ddmitry committed 2026-09-15 16:20:45 +03:00
commit 872b876be1
8 files changed
+514 -106

No files matched your search

+3 -3
View File
@@ -50,7 +50,7 @@
1. Понимает различие между `storage`, `catalog` и `compute`, и может объяснить роль `MinIO`, `PostgreSQL`, `Spark` и `Trino` в этом стенде. 1. Понимает различие между `storage`, `catalog` и `compute`, и может объяснить роль `MinIO`, `PostgreSQL`, `Spark` и `Trino` в этом стенде.
2. Умеет поднять локальный стенд, проверить его базовую работоспособность и использовать логи/веб-интерфейсы для первичной диагностики проблем. 2. Умеет поднять локальный стенд, проверить его базовую работоспособность и использовать логи/веб-интерфейсы для первичной диагностики проблем.
3. Умеет загрузить raw-данные, прочитать их в `Spark`, проверить схему и подготовить данные к дальнейшей обработке. 3. Умеет настроить Spark-сессию для кластера, объяснить происхождение настроек и повторить подключение в новом ядре; загрузить raw-данные, прочитать их и проверить схему.
4. Умеет создать и заполнить `Iceberg`-таблицу как основной табличный формат курса. 4. Умеет создать и заполнить `Iceberg`-таблицу как основной табличный формат курса.
5. Умеет реализовать простой поток `raw -> bronze -> silver` с базовыми трансформациями и сохранением воспроизводимости шагов. 5. Умеет реализовать простой поток `raw -> bronze -> silver` с базовыми трансформациями и сохранением воспроизводимости шагов.
6. Умеет читать одну и ту же таблицу из `Spark` и `Trino` и понимает, почему это возможно без копирования данных. 6. Умеет читать одну и ту же таблицу из `Spark` и `Trino` и понимает, почему это возможно без копирования данных.
@@ -90,7 +90,7 @@
* один основной табличный формат: `Iceberg`; * один основной табличный формат: `Iceberg`;
* работа с raw-данными; * работа с raw-данными;
* построение слоев `bronze` и `silver`; * построение слоев `bronze` и `silver`;
* чтение и запись через `Spark`; * базовая настройка сессии, чтение и запись через `Spark`;
* чтение через `Trino`; * чтение через `Trino`;
* базовое schema evolution; * базовое schema evolution;
* snapshot/time travel на прикладном уровне; * snapshot/time travel на прикладном уровне;
@@ -155,7 +155,7 @@
1. В составе курса есть стартовая документация до первого ноутбука, включая prerequisites, запуск стенда, вход в `Jupyter` и получение учебного data bundle. 1. В составе курса есть стартовая документация до первого ноутбука, включая prerequisites, запуск стенда, вход в `Jupyter` и получение учебного data bundle.
2. Студент проходит обязательные `checkpoints` по урокам. 2. Студент проходит обязательные `checkpoints` по урокам.
3. Студент самостоятельно выполняет набор практик на одном датасете, включая: 3. Студент самостоятельно выполняет набор практик на одном датасете, включая:
* загрузку raw-данных; * подключение к Spark и загрузку raw-данных;
* построение `bronze`; * построение `bronze`;
* построение `silver`; * построение `silver`;
* создание/чтение `Iceberg`-таблицы; * создание/чтение `Iceberg`-таблицы;
+16 -14
View File
@@ -69,7 +69,7 @@
| --- | --- | --- | | --- | --- | --- |
| `LO1`. Понимание различий между `storage`, `catalog`, `compute` | 1, 2, 6 | checkpoints модулей 2 и 6 | | `LO1`. Понимание различий между `storage`, `catalog`, `compute` | 1, 2, 6 | checkpoints модулей 2 и 6 |
| `LO2`. Умение поднять и диагностировать стенд | 1 | checkpoint модуля 1 | | `LO2`. Умение поднять и диагностировать стенд | 1 | checkpoint модуля 1 |
| `LO3`. Умение загрузить raw-данные и проверить схему | 3 | checkpoint модуля 3 | | `LO3`. Умение подключиться к Spark, загрузить raw-данные и проверить схему | 3 | checkpoint модуля 3 |
| `LO4`. Умение создать и заполнить `Iceberg`-таблицу | 4 | checkpoint модуля 4 | | `LO4`. Умение создать и заполнить `Iceberg`-таблицу | 4 | checkpoint модуля 4 |
| `LO5`. Умение собрать поток `raw -> bronze -> silver` | 4, 5, 8 | checkpoints модулей 5 и 8 | | `LO5`. Умение собрать поток `raw -> bronze -> silver` | 4, 5, 8 | checkpoints модулей 5 и 8 |
| `LO6`. Умение читать одну таблицу из `Spark` и `Trino` | 6, 8 | checkpoint модуля 6 и финальная практика | | `LO6`. Умение читать одну таблицу из `Spark` и `Trino` | 6, 8 | checkpoint модуля 6 и финальная практика |
@@ -124,25 +124,27 @@
**Рабочее название ноутбука:** `03_raw_ingest_and_first_read.ipynb` **Рабочее название ноутбука:** `03_raw_ingest_and_first_read.ipynb`
**Зачем нужен:** показать, как raw-данные попадают в стенд и как с ними безопасно начать работать. **Зачем нужен:** пройти путь от подключения к Spark до первого чтения и проверки исходных файлов Taxi.
**Ориентир по времени:** около 2 часов, из них 20–30 минут на подготовку Spark-сессии.
**Содержание:** **Содержание:**
* работа с заранее подготовленным локальным data bundle из onboarding-документации; * подключение к Spark-кластеру: сессия, контекст, driver и executors; настройки из файла, параметров запуска и кода;
* явная загрузка исходных `PARQUET`-файлов в raw-зону; * повторное использование сессии, остановка и перезапуск ядра; изменение настроек во время работы;
* чтение raw-данных через Spark; * подготовленный data bundle и загрузка исходных файлов в raw-зону MinIO;
* проверка схемы, типов и базового качества данных; * чтение через Spark, проверка схемы и типов;
* обсуждение того, почему raw лучше не "чинить на месте" и почему датасет не должен скачиваться "из интернета из ноутбука". * базовые метрики, аномалии и причины сохранять raw без изменений.
**Практика студента:** **Практика студента:**
* проверить состав локально полученного набора `NYC Taxi`; * заполнить шаблон сессии с подготовленным каталогом Iceberg; проверить подключение на собственной временной таблице и удалить ее;
* загрузить его в raw-зону `MinIO`; * изменить и восстановить часовой пояс, сравнить повторный вызов builder-а с новым запуском приложения;
* прочитать несколько месяцев `NYC Taxi`; * проверить состав data bundle и загрузить несколько месяцев Taxi в MinIO;
* проверить схему и посчитать базовые метрики; * прочитать поездки и справочник зон, проверить схему и посчитать метрики;
* найти простые аномалии или проблемные значения. * найти проблемные значения и объяснить, почему их сохраняют в raw.
**Checkpoint:** студент умеет загрузить учебный набор в raw-зону, прочитать raw-данные, проверить схему и объяснить, почему raw-слой нужен как воспроизводимая точка входа. **Checkpoint:** студент создает сессию для кластера, объясняет источники настроек и место выполнения заданий, умеет повторить подключение в новом ядре. Загружает учебный набор в raw-зону, читает его через Spark, проверяет схему и объясняет, зачем сохранять исходные данные.
### Модуль 4. Первая рабочая Iceberg-таблица и слой bronze ### Модуль 4. Первая рабочая Iceberg-таблица и слой bronze
@@ -256,7 +258,7 @@
1. Поднимает локальный стенд и диагностирует типовые проблемы на старте. 1. Поднимает локальный стенд и диагностирует типовые проблемы на старте.
2. Понимает различие между `storage`, `catalog` и `compute`. 2. Понимает различие между `storage`, `catalog` и `compute`.
3. Умеет загрузить raw-данные и прочитать их в Spark. 3. Подключается к Spark-кластеру, загружает raw-данные и проверяет их после чтения.
4. Умеет создать и заполнить Iceberg-таблицу. 4. Умеет создать и заполнить Iceberg-таблицу.
5. Умеет построить простой поток `raw -> bronze -> silver`. 5. Умеет построить простой поток `raw -> bronze -> silver`.
6. Умеет читать одну и ту же таблицу из `Spark` и `Trino`. 6. Умеет читать одну и ту же таблицу из `Spark` и `Trino`.
+16
View File
@@ -44,6 +44,22 @@
**Параллель с DWH:** в классическом DWH вычислитель один, поэтому проблема не возникает. **Параллель с DWH:** в классическом DWH вычислитель один, поэтому проблема не возникает.
**Где в курсе:** Модуль 6. **Где в курсе:** Модуль 6.
## Spark-сессия и выполнение
### SparkSession
Точка входа в DataFrame и Spark SQL: объект `spark` в ноутбуке. Builder задаёт параметры, а `getOrCreate()` создаёт сессию или возвращает уже работающую.
### SparkContext
Связывает Spark-приложение с вычислительными ресурсами. Через `spark.sparkContext` можно проверить фактические `master`, `appName` и `applicationId`. Повторный builder не заменяет работающий контекст.
### Driver и executors
Driver управляет приложением: выполняет код ноутбука, строит план и отправляет задания. Executors выполняют задания и читают данные. В нашем стенде driver работает в `jupyter`, executors запускаются на Spark worker-ах. Master выделяет приложению ресурсы worker-ов.
**Где в курсе:** Модуль 3.
## Структура Iceberg-таблицы ## Структура Iceberg-таблицы
### Namespace ### Namespace
+5
View File
@@ -64,6 +64,11 @@ docker compose down -v
или настройки в файле, перезапусти ядро ноутбука и снова выполни ячейки или настройки в файле, перезапусти ядро ноутбука и снова выполни ячейки
с импортами и созданием сессии. с импортами и созданием сессии.
Фактический master и имя приложения доступны через `spark.sparkContext.master`
и `.appName`. Текущие SQL-настройки можно прочитать через `spark.conf.get(...)`.
В Spark 3.5.1 `spark.sql.session.timeZone` меняется через `spark.conf.set(...)`
без остановки сессии. Для нового имени приложения нужен новый SparkContext.
Пример запуска smoke-скрипта: Пример запуска smoke-скрипта:
```bash ```bash
+330 -48
View File
@@ -7,15 +7,9 @@
"source": [ "source": [
"# Модуль 3. Raw-данные и первое чтение в Spark\n", "# Модуль 3. Raw-данные и первое чтение в Spark\n",
"\n", "\n",
"В этом модуле мы разбираем полный первый цикл работы с учебным датасетом в Lakehouse:\n", "В прошлом модуле мы проследили, как Spark работает с каталогом и файлами в MinIO. Теперь загрузим настоящие данные: поездки NYC Taxi и справочник зон.\n",
"\n", "\n",
"- проверяем заранее скачанный локальный data bundle;\n", "Сначала подготовим Spark-сессию и проверим подключение. Затем перенесем заранее скачанные файлы в raw-зону MinIO, прочитаем их через Spark и посмотрим на схему, пропуски и аномалии. Эти файлы останутся источником для таблиц, которые мы построим в следующих модулях.\n"
"- загружаем исходные файлы в raw-зону `MinIO`;\n",
"- читаем raw-данные через `Spark` по `s3a://` пути;\n",
"- смотрим схему, типы, базовые метрики и простые аномалии;\n",
"- фиксируем, почему raw-слой нужен как воспроизводимая неизменяемая точка входа.\n",
"\n",
"После этого ноутбука ты должен уметь объяснить, почему raw-данные не надо «чинить на месте» и почему их не стоит скачивать прямо из ноутбука.\n"
] ]
}, },
{ {
@@ -25,20 +19,319 @@
"source": [ "source": [
"## 0. Перед стартом\n", "## 0. Перед стартом\n",
"\n", "\n",
"Перед выполнением ноутбука:\n", "- Подними стенд по `START_HERE.md` и пройди модули 1 и 2.\n",
"- Скачай data bundle в `./data/nyc_taxi` по инструкции из `START_HERE.md`.\n",
"- Открой этот ноутбук в новом ядре, чтобы начать без сессии от предыдущих опытов.\n"
]
},
{
"cell_type": "markdown",
"id": "session-intro",
"metadata": {},
"source": [
"## 1. Как настроен Spark в нашем стенде\n",
"\n", "\n",
"- подними стенд по `START_HERE.md`;\n", "До сих пор мы пользовались готовым объектом `spark`. Разберем, как его создать и куда он отправляет задания: это понадобится при чтении файлов Taxi.\n",
"- пройди `notebooks/01_environment_and_smoke_test.ipynb`;\n",
"- желательно пройти `notebooks/02_lakehouse_mental_model.ipynb`;\n",
"- заранее скачай data bundle в `./data/nyc_taxi` по инструкции из `START_HERE.md`.\n",
"\n", "\n",
"В классическом `PostgreSQL / Greenplum` данные часто уже лежат внутри системы. В Lakehouse путь более явный: сначала исходные файлы должны попасть в object storage, и только потом мы начинаем поверх них читать, профилировать и строить таблицы.\n", "`SparkSession` дает доступ к DataFrame и SQL. Его `SparkContext` связывает приложение с вычислительными ресурсами. В обычном ядре нашего ноутбука используется один работающий контекст.\n",
"\n", "\n",
"| | Классический DWH | Lakehouse |\n", "| Участник | Что делает в стенде |\n",
"| --- | --- | --- |\n", "| --- | --- |\n",
"| Где стартует работа | Данные уже внутри СУБД | Исходные файлы сначала попадают в storage |\n", "| Driver | Работает в `jupyter`: выполняет Python-код, строит план и раздает задания |\n",
"| Первая практическая задача | `SELECT` из готовой таблицы | Загрузить raw и прочитать raw |\n", "| Master | Принимает подключение по `spark://spark-master:7077` и выделяет ресурсы worker-ов |\n",
"| Что важно не сломать | Таблицу / схему БД | Неизменяемый raw-слой |\n" "| Executors | Работают на `spark-worker-1` и `spark-worker-2`, читают данные и выполняют задания |\n",
"\n",
"В режиме `local[*]` задания выполняются в потоках на машине driver-а. Нам нужен кластер, поэтому проверим фактический master вместе с результатом вычисления.\n",
"\n",
"Создадим сессию. `.appName(...)` задает имя приложения в Spark UI, `.getOrCreate()` создает сессию или возвращает уже работающую. Адрес кластера пока возьмем из общего файла настроек.\n"
]
},
{
"cell_type": "code",
"id": "session-demo",
"metadata": {},
"source": [
"from pyspark.sql import SparkSession\n",
"\n",
"spark = (\n",
" SparkSession.builder\n",
" .appName(\"module-03-session-demo\")\n",
" .getOrCreate()\n",
")\n",
"spark.sparkContext.setLogLevel(\"ERROR\")\n",
"\n",
"print(\"Spark:\", spark.version)\n",
"print(\"Master:\", spark.sparkContext.master)\n",
"print(\"Приложение:\", spark.sparkContext.appName)\n",
"assert spark.sparkContext.master == \"spark://spark-master:7077\", \"Нужен Spark-кластер стенда\"\n",
"print(\"Сумма 0…9:\", spark.range(10).groupBy().sum(\"id\").first()[0])"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-config-explanation",
"metadata": {},
"source": [
"### Откуда взялось подключение\n",
"\n",
"Master мы не указали в коде: Spark прочитал его из `spark/spark-defaults.conf`. Docker подключает этот файл к Jupyter как `/opt/spark/conf/spark-defaults.conf`. Там же подготовлены каталог Iceberg и доступ к MinIO.\n",
"\n",
"| Настройка в файле | Для чего нужна |\n",
"| --- | --- |\n",
"| `spark.sql.extensions` | Добавляет команды Iceberg к Spark SQL |\n",
"| `spark.sql.catalog.lakehouse` | Подключает Spark-каталог с именем `lakehouse` |\n",
"| `spark.sql.catalog.lakehouse.catalog-impl` и `.uri` | Выбирает JDBC-каталог в PostgreSQL, где зарегистрированы таблицы |\n",
"| `spark.sql.catalog.lakehouse.warehouse` | Задает место для файлов таблиц: `s3a://lakehouse/warehouse` |\n",
"| `spark.hadoop.fs.s3a.endpoint` и `.path.style.access` | Направляют S3A-запросы в MinIO внутри Docker-сети |\n",
"\n",
"При запуске через `spark-submit` настройки можно передать параметрами `--master`, `--name` и `--conf`. В коде для этого есть `.master(...)`, `.appName(...)` и `.config(...)`. Для нового приложения обычный приоритет: код -> параметры запуска -> файл defaults. Параметры запуска driver-а надо задать до его старта; уже работающий контекст новый builder не перестраивает.\n",
"\n",
"В нашем Jupyter дополнительные `--master` и `--conf` при запуске не передаются. Вне стенда потребуются также библиотеки Iceberg, S3A и JDBC: их состав и конфигурация описаны в [справочнике](../docs/stack_reference.md#spark).\n",
"\n",
"Посмотрим, какие значения получила сессия. В файле есть и учетные данные, поэтому выводим только выбранные несекретные настройки. Для SQL используем `spark.conf`; в Hadoop-настройках префикс `spark.hadoop.` уже снят.\n"
]
},
{
"cell_type": "code",
"id": "session-config",
"metadata": {},
"source": [
"for key in (\n",
" \"spark.sql.extensions\",\n",
" \"spark.sql.catalog.lakehouse\",\n",
" \"spark.sql.catalog.lakehouse.catalog-impl\",\n",
" \"spark.sql.catalog.lakehouse.uri\",\n",
" \"spark.sql.catalog.lakehouse.warehouse\",\n",
"):\n",
" print(f\"{key} = {spark.conf.get(key)}\")\n",
"\n",
"hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()\n",
"for key in (\"fs.s3a.endpoint\", \"fs.s3a.path.style.access\"):\n",
" print(f\"{key} = {hadoop_conf.get(key)}\")"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-live-explanation",
"metadata": {},
"source": [
"### Изменение настройки в работающей сессии\n",
"\n",
"Некоторые настройки можно менять, пока приложение работает. Например, `spark.sql.session.timeZone` определяет часовой пояс SQL-сессии. Посмотрим, как одна и та же секунда Unix-времени отображается в UTC и Москве.\n",
"\n",
"В Spark 3.5.1 эта настройка изменяется через `spark.conf.set(...)`. Сохраним исходное значение и вернем его в `finally`, в том числе при ошибке. Файлы данных этот опыт не меняет.\n"
]
},
{
"cell_type": "code",
"id": "session-live-config",
"metadata": {},
"source": [
"timezone_key = \"spark.sql.session.timeZone\"\n",
"original_timezone = spark.conf.get(timezone_key)\n",
"assert spark.conf.isModifiable(timezone_key)\n",
"try:\n",
" for timezone in (\"UTC\", \"Europe/Moscow\"):\n",
" spark.conf.set(timezone_key, timezone)\n",
" value = spark.sql(\"SELECT from_unixtime(0) AS displayed_time\").first()[0]\n",
" print(timezone, value)\n",
"finally:\n",
" spark.conf.set(timezone_key, original_timezone)\n",
"\n",
"assert spark.conf.get(timezone_key) == original_timezone\n",
"print(\"Исходный часовой пояс восстановлен:\", original_timezone)"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-reuse-explanation",
"metadata": {},
"source": [
"### Повторный вызов builder-а\n",
"\n",
"Теперь попробуем сменить имя приложения на `module-03-raw-ingest-and-first-read`, с которым будем читать Taxi. Проверим имя у работающего контекста: `spark.sparkContext.appName`.\n"
]
},
{
"cell_type": "code",
"id": "session-reuse",
"metadata": {},
"source": [
"demo_context = spark.sparkContext\n",
"demo_name = demo_context.appName\n",
"same_spark = (\n",
" SparkSession.builder\n",
" .appName(\"module-03-raw-ingest-and-first-read\")\n",
" .getOrCreate()\n",
")\n",
"assert same_spark is spark\n",
"assert same_spark.sparkContext is demo_context\n",
"assert same_spark.sparkContext.appName == demo_name\n",
"print(\"То же приложение:\", same_spark.sparkContext.appName)"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-stop-explanation",
"metadata": {},
"source": [
"Имя осталось прежним: `getOrCreate()` вернул ту же сессию с тем же контекстом. Повтор ячейки сохраняет и переменные Python, и работающее приложение.\n",
"\n",
"Чтобы создать приложение с новым именем, вызовем `spark.stop()`. Этот вызов останавливает сессию и ее контекст. Переменные Python останутся, но старые Spark DataFrame использовать уже нельзя. Файлы MinIO и таблицы каталога сохранятся.\n",
"\n",
"Для смены master или настроек в общем файле перезапускай ядро: одной остановки сессии недостаточно для всех параметров запуска. Перезапуск удалит переменные Python, и тогда придется повторить импорты, создание сессии и чтение DataFrame.\n"
]
},
{
"cell_type": "code",
"id": "session-stop",
"metadata": {},
"source": [
"spark.stop()"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-exercise-explanation",
"metadata": {},
"source": [
"## 2. Подключись к Spark самостоятельно\n",
"\n",
"Создай сессию, с которой продолжишь работу с Taxi. Заполни в следующей ячейке адрес кластера, имя `module-03-raw-ingest-and-first-read` и цепочку `.master(...)`, `.appName(...)`, `.getOrCreate()`.\n",
"\n",
"Настройки каталога возьми из общего файла. Адрес кластера в этот раз задай явно через `.master(...)`; остальные ноутбуки берут его из того же файла.\n",
"\n",
"Пока пропуски не заполнены, шаблон останавливается на `assert`. Выполни заполненный шаблон и проверку подключения дважды: имя и `applicationId` должны остаться теми же.\n"
]
},
{
"cell_type": "code",
"id": "session-exercise",
"metadata": {},
"source": [
"from pyspark.sql import SparkSession\n",
"\n",
"CLUSTER_MASTER = ...\n",
"APP_NAME = ...\n",
"assert CLUSTER_MASTER == \"spark://spark-master:7077\", \"Впиши адрес кластера\"\n",
"assert APP_NAME == \"module-03-raw-ingest-and-first-read\", \"Впиши имя приложения\"\n",
"\n",
"# Замени ... цепочкой вызовов SparkSession.builder.\n",
"spark = ...\n",
"assert isinstance(spark, SparkSession), \"Создай SparkSession через builder\"\n",
"spark.sparkContext.setLogLevel(\"ERROR\")"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-solution",
"metadata": {},
"source": [
"<details>\n",
"<summary>Контрольное решение (после своей попытки)</summary>\n",
"\n",
"```python\n",
"from pyspark.sql import SparkSession\n",
"\n",
"CLUSTER_MASTER = \"spark://spark-master:7077\"\n",
"APP_NAME = \"module-03-raw-ingest-and-first-read\"\n",
"spark = (\n",
" SparkSession.builder\n",
" .master(CLUSTER_MASTER)\n",
" .appName(APP_NAME)\n",
" .getOrCreate()\n",
")\n",
"spark.sparkContext.setLogLevel(\"ERROR\")\n",
"```\n",
"\n",
"</details>"
]
},
{
"cell_type": "code",
"id": "session-connection-check",
"metadata": {},
"source": [
"assert spark.sparkContext.master == CLUSTER_MASTER\n",
"assert spark.sparkContext.appName == APP_NAME\n",
"print(\"Master:\", spark.sparkContext.master)\n",
"print(\"Приложение:\", spark.sparkContext.appName)\n",
"print(\"ID приложения:\", spark.sparkContext.applicationId)"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-table-explanation",
"metadata": {},
"source": [
"### Запись и чтение через каталог\n",
"\n",
"Проверим, что сессия может работать с каталогом и MinIO. Следующая ячейка создаст собственную маленькую Iceberg-таблицу, прочитает из нее числа `0` и `1` и удалит ее в `finally`. Она работает и после удаления таблиц из модуля 2.\n",
"\n",
"У таблицы уникальное имя. `DROP TABLE ... PURGE` удаляет ее запись в каталоге, данные и метаданные; в MinIO могут остаться пустые маркеры `data/` и `metadata/`.\n",
"\n",
"Дождись завершения ячейки перед перезапуском ядра: при принудительной остановке процесса `finally` не выполнится. Если прервал ее, сохрани напечатанное имя, чтобы удалить эту таблицу после нового подключения.\n"
]
},
{
"cell_type": "code",
"id": "session-table-check",
"metadata": {},
"source": [
"from uuid import uuid4\n",
"\n",
"session_table = f\"lakehouse.default.session_practice_{uuid4().hex}\"\n",
"print(\"Временная таблица:\", session_table)\n",
"try:\n",
" spark.range(2).writeTo(session_table).using(\"iceberg\").create()\n",
" session_rows = [row.id for row in spark.table(session_table).orderBy(\"id\").collect()]\n",
" assert session_rows == [0, 1]\n",
" print(\"Прочитано через Iceberg:\", session_rows)\n",
"finally:\n",
" spark.sql(f\"DROP TABLE IF EXISTS {session_table} PURGE\")\n",
"\n",
"assert not spark.catalog.tableExists(session_table)\n",
"print(\"Временная таблица удалена\")"
],
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "session-checkpoint",
"metadata": {},
"source": [
"### Проверь запуск в новом ядре\n",
"\n",
"В выводе должны быть master `spark://spark-master:7077`, заданное имя приложения, строки `[0, 1]` и сообщение об удалении таблицы.\n",
"\n",
"Перезапусти ядро и повтори заполненный шаблон, проверку подключения и ячейку с таблицей. `applicationId` будет новым, остальные результаты должны совпасть. Импорт `SparkSession` уже есть в шаблоне, поэтому демонстрацию повторять не нужно.\n",
"\n",
"С этим подключением перейдем к файлам Taxi. Оставь сессию работать.\n"
]
},
{
"cell_type": "markdown",
"id": "raw-preparation",
"metadata": {},
"source": [
"## 3. Проверка data bundle\n",
"\n",
"Подготовим MinIO-клиент для загрузки файлов и пути к данным. Docker подключает каталог `data/` с хоста к Jupyter как `/opt/data/`.\n"
] ]
}, },
{ {
@@ -63,10 +356,6 @@
"RAW_URI = f\"s3a://{BUCKET_NAME}/{RAW_PREFIX}\"\n", "RAW_URI = f\"s3a://{BUCKET_NAME}/{RAW_PREFIX}\"\n",
"ZONE_LOOKUP_RAW_URI = f\"{RAW_URI}taxi_zone_lookup.csv\"\n", "ZONE_LOOKUP_RAW_URI = f\"{RAW_URI}taxi_zone_lookup.csv\"\n",
"\n", "\n",
"spark = SparkSession.builder .appName(\"module-03-raw-ingest-and-first-read\") .getOrCreate()\n",
"\n",
"spark.sparkContext.setLogLevel(\"ERROR\")\n",
"\n",
"s3 = boto3.client(\n", "s3 = boto3.client(\n",
" \"s3\",\n", " \"s3\",\n",
" endpoint_url=\"http://minio:9000\",\n", " endpoint_url=\"http://minio:9000\",\n",
@@ -127,7 +416,7 @@
"\n", "\n",
"\n", "\n",
"print(\"Spark и MinIO-клиент готовы\")\n", "print(\"Spark и MinIO-клиент готовы\")\n",
"print(f\"RAW URI: {RAW_URI}\")\n" "print(f\"RAW URI: {RAW_URI}\")"
] ]
}, },
{ {
@@ -135,15 +424,7 @@
"id": "d4d8f302", "id": "d4d8f302",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 1. Проверка data bundle\n", "Файлы должны быть скачаны заранее по `START_HERE.md`. Здесь проверим состав набора, прежде чем отправлять его в MinIO. Так во время урока нам не понадобится внешняя сеть, а при повторном прохождении мы будем читать те же исходные файлы.\n"
"\n",
"Сначала убеждаемся, что нужные файлы уже лежат внутри Jupyter-контейнера по пути `/opt/data/nyc_taxi`.\n",
"\n",
"Почему не скачиваем данные прямо здесь:\n",
"\n",
"- так модуль остаётся воспроизводимым и не зависит от сетевого доступа внутри ноутбука;\n",
"- onboarding и подготовка окружения остаются в `START_HERE.md`, а практика в ноутбуке занимается только Lakehouse-операциями;\n",
"- студент видит границу ответственности: сначала подготовить bundle, потом работать с raw-слоем.\n"
] ]
}, },
{ {
@@ -198,7 +479,7 @@
"id": "dabd90f2", "id": "dabd90f2",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 2. Загрузка в raw-зону MinIO\n", "## 4. Загрузка в raw-зону MinIO\n",
"\n", "\n",
"Raw-слой в Lakehouse похож на воспроизводимый входной `stg`-контур: мы сохраняем исходные файлы как есть, под оригинальными именами, не переписывая их содержимое.\n", "Raw-слой в Lakehouse похож на воспроизводимый входной `stg`-контур: мы сохраняем исходные файлы как есть, под оригинальными именами, не переписывая их содержимое.\n",
"\n", "\n",
@@ -296,7 +577,7 @@
"id": "696d6adc", "id": "696d6adc",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 3. Первое чтение raw PARQUET через Spark\n", "## 5. Первое чтение raw PARQUET через Spark\n",
"\n", "\n",
"Теперь читаем raw-файлы напрямую из `MinIO` по `s3a://` пути. Это и есть первая важная практика Lakehouse: вычислитель (`Spark`) работает поверх файлов в object storage, а не только поверх локальной файловой системы или таблиц внутри одной СУБД.\n", "Теперь читаем raw-файлы напрямую из `MinIO` по `s3a://` пути. Это и есть первая важная практика Lakehouse: вычислитель (`Spark`) работает поверх файлов в object storage, а не только поверх локальной файловой системы или таблиц внутри одной СУБД.\n",
"\n", "\n",
@@ -305,7 +586,7 @@
"- `spark.read.parquet(\"s3a://...\")` читает raw-файлы напрямую по физическому пути;\n", "- `spark.read.parquet(\"s3a://...\")` читает raw-файлы напрямую по физическому пути;\n",
"- `spark.table(\"lakehouse....\")` читает уже зарегистрированную управляемую таблицу через каталог.\n", "- `spark.table(\"lakehouse....\")` читает уже зарегистрированную управляемую таблицу через каталог.\n",
"\n", "\n",
"Сейчас мы ещё не создавали `Iceberg`-таблицу, поэтому работаем именно через `spark.read.parquet(...)`. К `spark.table(...)` вернёмся в Модуле 4.\n" "Для проверки подключения мы читали маленькую таблицу по имени через `spark.table(...)`. Файлы Taxi пока не зарегистрированы в каталоге: прочитаем их по пути через `spark.read.parquet(...)`. Таблицу поездок создадим в модуле 4."
] ]
}, },
{ {
@@ -372,7 +653,7 @@
"id": "9701728a", "id": "9701728a",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 4. Схема, типы и null-значения\n", "## 6. Схема, типы и null-значения\n",
"\n", "\n",
"Следующий шаг после первого чтения: понять, какие колонки пришли, какие у них типы и где уже на raw-слое встречаются `null`.\n", "Следующий шаг после первого чтения: понять, какие колонки пришли, какие у них типы и где уже на raw-слое встречаются `null`.\n",
"\n", "\n",
@@ -435,7 +716,7 @@
"id": "69e6b77e", "id": "69e6b77e",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 5. Базовые метрики и аномалии\n", "## 7. Базовые метрики и аномалии\n",
"\n", "\n",
"Теперь быстро профилируем данные и ищем несколько очевидных проблемных паттернов:\n", "Теперь быстро профилируем данные и ищем несколько очевидных проблемных паттернов:\n",
"\n", "\n",
@@ -554,7 +835,7 @@
"id": "095c9ac6", "id": "095c9ac6",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 6. Самостоятельное задание\n", "## 8. Самостоятельное задание\n",
"\n", "\n",
"Используй raw-данные, которые уже лежат в `MinIO`:\n", "Используй raw-данные, которые уже лежат в `MinIO`:\n",
"\n", "\n",
@@ -612,7 +893,7 @@
"id": "e23dfe47", "id": "e23dfe47",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 7. Почему raw-слой важен\n", "## 9. Почему raw-слой важен\n",
"\n", "\n",
"Raw-слой полезен не потому, что в нём удобно жить постоянно, а потому что он даёт устойчивую исходную точку. Если завтра поменяются правила очистки, дедупликации или нормализации, мы не обязаны заново скачивать источник и гадать, что именно уже успели «подправить». Мы просто перечитываем raw и перестраиваем downstream-слои.\n", "Raw-слой полезен не потому, что в нём удобно жить постоянно, а потому что он даёт устойчивую исходную точку. Если завтра поменяются правила очистки, дедупликации или нормализации, мы не обязаны заново скачивать источник и гадать, что именно уже успели «подправить». Мы просто перечитываем raw и перестраиваем downstream-слои.\n",
"\n", "\n",
@@ -628,16 +909,17 @@
"id": "8849c73f", "id": "8849c73f",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 8. Checkpoint\n", "## 10. Checkpoint\n",
"\n", "\n",
"Проверь себя:\n", "Проверь себя:\n",
"\n", "\n",
"1. Можешь ли ты показать файлы raw-зоны в `MinIO` через код или `MinIO Console`?\n", "1. К какому master подключена сессия? Какие настройки ты задал в коде, а какие она получила из общего файла?\n",
"2. Сколько колонок в схеме `Yellow Taxi` и какие типы там встречаются?\n", "2. Чем SparkSession отличается от SparkContext? Где работают driver и executors?\n",
"3. Какие колонки содержат `null` и почему это не обязательно баг загрузки?\n", "3. Почему часовой пояс изменился сразу, а для нового имени приложения понадобилось остановить сессию? Что придется повторить после перезапуска ядра?\n",
"4. Зачем нужен raw-слой, если дальше всё равно будут трансформации?\n", "4. Можешь ли ты показать загруженные raw-файлы в MinIO и прочитать их через Spark?\n",
"5. Почему мы не скачиваем датасет прямо из ноутбука?\n", "5. Сколько колонок в Yellow Taxi, какие у них типы и где встречаются `null`? Почему пропуски не обязательно означают ошибку загрузки?\n",
"6. Что произойдёт, если источник завтра поменяет формат файлов?\n" "6. Зачем сохранять raw-слой без изменений, если дальше будут трансформации?\n",
"7. Почему мы скачали датасет до запуска ноутбука? Что потребуется проверить, если источник изменит формат файлов?\n"
] ]
}, },
{ {
@@ -645,7 +927,7 @@
"id": "371c4c9d", "id": "371c4c9d",
"metadata": {}, "metadata": {},
"source": [ "source": [
"## 9. Завершение\n", "## 11. Завершение\n",
"\n", "\n",
"Мы намеренно не удаляем raw-данные из `MinIO`. В Модуле 4 они понадобятся для построения первого управляемого `bronze`-слоя на `Iceberg`.\n" "Мы намеренно не удаляем raw-данные из `MinIO`. В Модуле 4 они понадобятся для построения первого управляемого `bronze`-слоя на `Iceberg`.\n"
] ]
+5 -17
View File
@@ -1,6 +1,6 @@
# План развития курса «Lakehouse без магии» # План развития курса «Lakehouse без магии»
Статус: подключение Spark исправлено; следующая работа — практика настройки сессии. Статус: ближайшие доработки завершены; ожидается обратная связь по оставшимся модулям.
Последнее обновление: 2026-09-15. Последнее обновление: 2026-09-15.
Здесь собираем отзывы студентов, выбираем следующие улучшения и объясняем их порядок. Подробные задания, критерии приёмки и актуальный статус исполнения хранятся в [Gitea Issues](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues). Здесь собираем отзывы студентов, выбираем следующие улучшения и объясняем их порядок. Подробные задания, критерии приёмки и актуальный статус исполнения хранятся в [Gitea Issues](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues).
@@ -36,26 +36,13 @@ Iceberg-таблицы. Результаты проверок и их огран
### 2. Добавить практику «Как настроен Spark в нашем стенде» ### 2. Добавить практику «Как настроен Spark в нашем стенде»
**Статус:** запланировано; можно приступать после завершения #1. **Статус:** выполнено, [MR #4](https://git.dementev.space/ddmitry/mini-lakehouse-lab/pulls/4).
**Задача:** [#2 — практика базовой настройки SparkSession](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/2). **Задача:** [#2 — практика базовой настройки SparkSession](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/2).
Практику разместить в начале модуля 3. Ориентир — 20–30 минут; фактическую длительность проверить при прохождении. В модуле 3 студент готовит Spark-сессию, проверяет подключение и затем переходит к загрузке и чтению Taxi. Содержание и checkpoint описаны в [программе](../docs/course_program.md#модуль-3-raw-данные-и-первое-чтение-в-spark).
Студент должен научиться: Код проверен на стенде, включая перезапуск ядра и полный raw-ingest. Замеры и ограничения проверки сохранены в [отчете](issue-2-spark-session-validation.md).
- объяснять роли `SparkSession`, `SparkContext`, driver и executors на примере стенда;
- читать код с `master`, `appName` и `getOrCreate`;
- находить источник настройки: конфигурационный файл, параметры запуска или код;
- проверять фактический master и выбранные настройки каталога Iceberg и MinIO;
- различать настройки, меняемые во время работы, и настройки, требующие нового запуска;
- понимать, что повторное выполнение ячейки может использовать уже существующую сессию.
**Практика и checkpoint:** студент заполняет небольшой шаблон создания сессии, задаёт подключение к кластеру и имя приложения, использует подготовленные настройки каталога, проверяет фактическое подключение и читает небольшую учебную Iceberg-таблицу. Таблица должна создаваться самой практикой: демонстрационные таблицы модуля 2 могут быть удалены при его завершении. Студент объясняет, какие настройки он задал сам, какие предоставил стенд и что потребуется повторить в новом ядре.
При реализации обновить learning outcomes и scope в PRD, содержание и checkpoint модуля 3 в программе, ноутбук и связанные справочные материалы. Сохранить структуру из восьми модулей.
**Граница темы:** базовая настройка подключения и жизненный цикл сессии. Подробный подбор памяти, числа executors, shuffle и GC остаётся за рамками этой доработки.
### 3. Собрать замечания по оставшейся части курса ### 3. Собрать замечания по оставшейся части курса
@@ -81,3 +68,4 @@ Iceberg-таблицы. Результаты проверок и их огран
## Выпущенные улучшения ## Выпущенные улучшения
- [#1](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/1): все восемь ноутбуков подключаются к Spark-кластеру при создании сессии в новом процессе; настройки собраны в `spark/spark-defaults.conf`. - [#1](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/1): все восемь ноутбуков подключаются к Spark-кластеру при создании сессии в новом процессе; настройки собраны в `spark/spark-defaults.conf`.
- [#2](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/2): модуль 3 включает самостоятельное подключение к Spark перед загрузкой и чтением Taxi.
+95
View File
@@ -0,0 +1,95 @@
# Проверка практики SparkSession в модуле 3
Дата: 2026-09-15.
Задача: https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/2.
База: `00942c2`, результат закрытой [#1](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/1) прочитан вместе с комментариями.
## Что проверялось
Фактические ячейки `notebooks/03_raw_ingest_and_first_read.ipynb` выполнялись
через `jupyter_client.KernelManager` в контейнере `jupyter`, Spark 3.5.1.
Менеджер запускал настоящее IPython-ядро и перезапускал его через
`restart_kernel(now=True)`. Проверка не подменяла создание сессии общим helper-ом.
Порядок воспроизведения:
1. Поднять стенд и получить стандартный bundle по `START_HERE.md`.
2. В новом ядре выполнить демонстрацию, вывод настроек, опыт с часовым поясом,
повторный builder и остановку сессии.
3. Заполнить три пропуска шаблона, выполнить его дважды и сравнить контекст
и applicationId. Выполнить опубликованное контрольное решение
(при проверке извлекалось из блока `session-solution`).
4. Выполнить проверку master/appName и дважды создать, прочитать и удалить
временную таблицу.
5. Перезапустить ядро. Убедиться, что переменные `spark` и `CLUSTER_MASTER`
отсутствуют; выполнить контрольное решение, проверку подключения и таблицы.
6. Продолжить исходный raw-ingest до `spark.stop()`, заполнив самостоятельные
упражнения с lookup, top-5 локаций и ещё одной аномалией.
7. Отдельно проверить заполненный шаблон в новом ядре и повторную загрузку
уже существующих raw-файлов.
## Результаты
| Проверка | Результат |
| --- | --- |
| Демонстрационная сессия | Master `spark://spark-master:7077`, appName `module-03-session-demo`, сумма 0…9 = 45 |
| Настройки каталога | `SparkCatalog`, `JdbcCatalog`, URI PostgreSQL и warehouse соответствуют общему файлу; Hadoop endpoint `http://minio:9000`, path-style `true` |
| Изменяемая настройка | `spark.sql.session.timeZone` допускает изменение; Unix-время 0 отображается как `1970-01-01 00:00:00` в UTC и `1970-01-01 03:00:00` в Москве |
| Восстановление | В `finally` восстановлено исходное `Etc/UTC`, проверено через assert |
| Повторный builder с другим appName | Тот же объект сессии и контекста, фактическое имя остаётся `module-03-session-demo` |
| Создание после stop | Новое имя `module-03-raw-ingest-and-first-read`, master кластерный |
| Повторение шаблона и контрольного решения | Контекст и applicationId не меняются |
| Заполненный шаблон в отдельном новом ядре | Master и appName верны, распределённая сумма 0…9 = 45 |
| Перезапуск ядра | Переменные исчезли, сессия создана заново; ID изменился с `app-20260915125347-0001` на `app-20260915125511-0002` |
| Iceberg до и после перезапуска | Три независимые таблицы с уникальными именами; каждый раз прочитаны `[0, 1]`, после PURGE таблица отсутствует в каталоге |
| Raw-ingest | Январь: 2 964 624 строки; январь–март: 9 554 778 строк, 19 колонок |
| Профилирование | Схема, null-доли, describe и четыре проверки аномалий выполнены без ошибок |
| Самостоятельное задание raw | Lookup: 265 строк; подсчёт зон по Borough и top-5 pickup с названиями зон выполнены; найдено 180 поездок с отрицательной длительностью |
| Повторная загрузка raw в отдельном ядре | Все четыре файла пропущены с `skipped_same_size`; снимки всех объектов и каталога до/после совпали |
Код исходного raw-ingest после ячейки подготовки не изменён, что проверено
сравнением ячеек с базовым коммитом. Из подготовки убрано только создание сессии:
она теперь приходит из новой практики.
## Время и ограничения
- Машинный прогон блока настройки: **179,27 с**. Сюда вошли дополнительные
повторы шаблона и таблицы, а также настоящий перезапуск ядра.
- Исходная часть модуля, включая заполненные самостоятельные упражнения:
**70,41 с**. Полный основной прогон: **250,02 с** (около 4 минут 10 секунд).
- Запуск контейнеров и скачивание bundle в замер не входят. Это время выполнения
кода; чтение объяснений, самостоятельное заполнение и ответы студента не измерялись.
- Ориентир новой практики **20–30 минут**, модуля целиком **около 2 часов**.
Общую оценку **12–15 часов** оставили предварительной; машинного замера
недостаточно для её уточнения. Нужен самостоятельный проход студента.
- Кнопки Jupyter/VS Code, отображение раскрывающегося решения и UI сервисов
не проверялись. Перезапуск выполнен менеджером ядра через API, а не кнопкой.
- Остальные семь ноутбуков и Trino smoke не выполнялись: runtime-конфигурация
не менялась, проверялся модуль 3.
## Сохранность данных
До основного прогона в каталоге было **0 таблиц**, в bucket — **4 объекта**.
После прогона исходные объекты сохранили ETag и размер, состав каталога
не изменился. Это проверка на имевшемся состоянии стенда, а не на копии
заполненного студентом курса.
Добавлены четыре исходных файла Taxi в `raw/nyc_taxi/`, как предусмотрено уроком;
они оставлены для модуля 4. Локальный bundle сохранён в игнорируемом `data/`.
Файлы данных и метаданных трёх временных таблиц удалены через PURGE.
Hadoop S3A оставил шесть нулевых маркеров `data/` и `metadata/`;
при завершении проверки удалены только эти шесть зафиксированных объектов.
В трёх префиксах практики осталось 0 объектов. Volumes не сбрасывались.
Поднятый для проверки стенд остановлен через `docker compose down` без `-v`.
## Статические проверки
- `git diff --check`;
- `docker compose config --quiet`;
- проверка ноутбука через `nbformat.validate` и синтаксиса Python-ячеек через AST;
- относительные ссылки изменённых документов и ноутбука;
- сохранение восьми модулей и маршрута README → START_HERE → программа → ноутбук.
После редакторской правки по ревью повторно проверены ссылки, нумерация разделов
и синтаксис Python. Все исполняемые ячейки, их порядок и контрольное решение
совпадают с проверенной версией. Стенд повторно не запускался.
+44 -24
View File
@@ -1,16 +1,22 @@
# Модуль 3. Raw-данные и первое чтение в Spark # Модуль 3. Raw-данные и первое чтение в Spark
**Статус:** `Ready for validation` **Статус:** `Done`
**Последнее обновление:** `2026-03-07` **Последнее обновление:** `2026-09-15`
**Задача:** [#2](https://git.dementev.space/ddmitry/mini-lakehouse-lab/issues/2).
Результаты прогона и ограничения: [проверка #2](issue-2-spark-session-validation.md).
## Цель ## Цель
Научить студента полному циклу работы с raw-данными в Lakehouse: доставка заранее скачанного PARQUET-датасета в raw-зону MinIO, чтение raw-данных через Spark, проверка схемы, типов и базового качества. Студент должен понять, зачем нужен raw-слой и почему raw-данные не следует «чинить на месте». Научить студента подключаться к Spark-кластеру и работать с raw-данными: загружать заранее скачанный PARQUET-датасет в MinIO, читать его через Spark и проверять схему, типы и базовое качество. Студент должен понять, зачем нужен raw-слой и почему raw-данные не следует «чинить на месте».
## Результат для студента ## Результат для студента
После прохождения модуля студент: После прохождения модуля студент:
- создает сессию для кластера с подготовленным каталогом, проверяет настройки и понимает их источники;
- различает повторный вызов builder-а, остановку сессии и перезапуск ядра;
- понимает, как локальный data bundle попадает в raw-зону MinIO; - понимает, как локальный data bundle попадает в raw-зону MinIO;
- умеет загружать PARQUET-файлы в MinIO программно через `boto3`; - умеет загружать PARQUET-файлы в MinIO программно через `boto3`;
- умеет читать raw-данные из MinIO через Spark (`spark.read.parquet` по `s3a://` пути); - умеет читать raw-данные из MinIO через Spark (`spark.read.parquet` по `s3a://` пути);
@@ -81,61 +87,73 @@ volumes:
## Структура ноутбука ## Структура ноутбука
### Секция 0: Введение ### Секция 0: Перед стартом
- **[md]** Заголовок, цели модуля, prerequisite (data bundle из `START_HERE.md`). Таблица-сравнение: «Классический DWH: данные уже внутри СУБД» vs «Lakehouse: данные нужно явно доставить в storage». - **[md]** Связь с модулем 2, цели работы с Taxi и условия старта: работающий стенд, скачанный data bundle, новое ядро.
### Секция 1: Проверка data bundle ### Секция 1: Как настроен Spark в нашем стенде
- **[md]** SparkSession, SparkContext, driver и executors, local и кластер; источники настроек и подготовленный каталог.
- **[code]** Создание сессии, фактический master и несекретные настройки. Изменение часового пояса с восстановлением. Повторный builder сохраняет контекст; `stop()` позволяет создать приложение с новым именем.
### Секция 2: Подключись к Spark самостоятельно
- **[code]** Шаблон подключения и контрольное решение. Проверка master и appName, запись и чтение `[0, 1]` из собственной временной таблицы, удаление в `finally`.
- **[md]** После удаления таблицы перезапустить ядро, повторить заполненный шаблон и проверки. Оставить сессию для чтения Taxi.
### Секция 3: Проверка data bundle
- **[code]** Импорты, MinIO-клиент, пути и вспомогательные функции для работы с файлами.
- **[md]** Зачем проверяем; почему не скачиваем из интернета в ноутбуке (воспроизводимость, сетевые ограничения, разделение ответственности onboarding vs практика). - **[md]** Зачем проверяем; почему не скачиваем из интернета в ноутбуке (воспроизводимость, сетевые ограничения, разделение ответственности onboarding vs практика).
- **[code]** `os.listdir("/opt/data/nyc_taxi")`, вывод файлов с размерами, assert на наличие PARQUET-файлов и `taxi_zone_lookup.csv` — оба нужны для полного прохождения модуля (включая самостоятельное задание в Секции 6). Понятное сообщение с отсылкой к `START_HERE.md` при отсутствии любого из компонентов. - **[code]** `os.listdir("/opt/data/nyc_taxi")`, вывод файлов с размерами, assert на наличие PARQUET-файлов и `taxi_zone_lookup.csv` — оба нужны для полного прохождения модуля (включая самостоятельное задание в Секции 8). Понятное сообщение с отсылкой к `START_HERE.md` при отсутствии любого из компонентов.
### Секция 2: Загрузка в raw-зону MinIO ### Секция 4: Загрузка в raw-зону MinIO
- **[md]** Что такое raw-зона, параллель с stg-слоем в Greenplum, конвенция путей, почему raw не чинится «на месте». - **[md]** Что такое raw-зона, параллель с stg-слоем в Greenplum, конвенция путей, почему raw не чинится «на месте».
- **[code]** Демонстрация: перед загрузкой проверяем через `s3.head_object()`, существует ли уже объект с таким ключом. Три ветки: (1) объект не существует — загружаем через `boto3.upload_file()`; (2) объект существует и размер совпадает с локальным файлом — пропускаем с пометкой «already exists, skipping»; (3) объект существует, но размер отличается — **fail fast** с понятной ошибкой (raise), без перезаписи. Это обеспечивает идемпотентность и подкрепляет принцип «raw не перезаписывается». В markdown-пояснении перед ячейкой явно объяснить: повторный запуск безопасен, уже загруженные файлы не перетираются, а конфликт размеров сигнализирует о проблеме с данными. - **[code]** Демонстрация: перед загрузкой проверяем через `s3.head_object()`, существует ли уже объект с таким ключом. Три ветки: (1) объект не существует — загружаем через `boto3.upload_file()`; (2) объект существует и размер совпадает с локальным файлом — пропускаем с пометкой «already exists, skipping»; (3) объект существует, но размер отличается — **fail fast** с понятной ошибкой (raise), без перезаписи. Это обеспечивает идемпотентность и подкрепляет принцип «raw не перезаписывается». В markdown-пояснении перед ячейкой явно объяснить: повторный запуск безопасен, уже загруженные файлы не перетираются, а конфликт размеров сигнализирует о проблеме с данными.
- **[code]** Верификация: `s3.list_objects_v2()` — листинг raw-зоны. - **[code]** Верификация: `s3.list_objects_v2()` — листинг raw-зоны.
- **[md]** Данные теперь в storage. Spark может читать их напрямую по `s3a://` пути; для Trino потребуется зарегистрированная таблица (это будет в Модулях 4 и 6). - **[md]** Данные теперь в storage. Spark может читать их напрямую по `s3a://` пути; для Trino потребуется зарегистрированная таблица (это будет в Модулях 4 и 6).
### Секция 3: Первое чтение в Spark ### Секция 5: Первое чтение в Spark
- **[md]** Разница `spark.read.parquet("s3a://...")` (raw-файлы) vs `spark.table(...)` (управляемая Iceberg-таблица). Мы пока на этапе raw, Iceberg-таблица появится в Модуле 4. - **[md]** Разница `spark.read.parquet("s3a://...")` (raw-файлы) vs `spark.table(...)` (управляемая Iceberg-таблица). Проверочная таблица удалена; таблицу поездок создаем в Модуле 4.
- **[code]** Создание SparkSession (паттерн из Module 2 — адрес кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`). - **[code]** Чтение через сессию, созданную в секции 2; повторная инициализация не нужна.
- **[code]** Чтение одного файла: `spark.read.parquet("s3a://lakehouse/raw/nyc_taxi/yellow_tripdata_2024-01.parquet")`. - **[code]** Чтение одного файла: `spark.read.parquet("s3a://lakehouse/raw/nyc_taxi/yellow_tripdata_2024-01.parquet")`.
- **[code]** Чтение всех файлов wildcard: `yellow_tripdata_*.parquet` (без привязки к конкретному году в wildcard, чтобы код работал с любыми скачанными файлами). - **[code]** Чтение всех файлов wildcard: `yellow_tripdata_*.parquet` (без привязки к конкретному году в wildcard, чтобы код работал с любыми скачанными файлами).
### Секция 4: Проверка схемы ### Секция 6: Проверка схемы
- **[md]** Зачем проверять схему: в классическом DWH схема задана DDL, в raw — вычитывается из файлов. Что может пойти не так: разные типы в разных файлах, неожиданные nullable-поля. - **[md]** Зачем проверять схему: в классическом DWH схема задана DDL, в raw — вычитывается из файлов. Что может пойти не так: разные типы в разных файлах, неожиданные nullable-поля.
- **[code]** `.printSchema()`. - **[code]** `.printSchema()`.
- **[code]** Колонки и типы в удобном формате (`.dtypes`). - **[code]** Колонки и типы в удобном формате (`.dtypes`).
- **[code]** Null-анализ ключевых колонок (`VendorID`, `tpep_pickup_datetime`, `passenger_count`, `trip_distance`, `total_amount`) — количество и доля null в каждой. - **[code]** Null-анализ ключевых колонок (`VendorID`, `tpep_pickup_datetime`, `passenger_count`, `trip_distance`, `total_amount`) — количество и доля null в каждой.
- **[md]** Обсуждение: null в raw — норма, не исправляем на месте. Очистка — задача bronze/silver слоев (Модули 4-5). - **[md]** Обсуждение: null в raw — норма, не исправляем на месте. Очистка — задача bronze/silver слоев (Модули 4-5).
### Секция 5: Базовые метрики и аномалии ### Секция 7: Базовые метрики и аномалии
- **[md]** Цель: быстро понять диапазоны значений, выявить очевидные проблемы. - **[md]** Цель: быстро понять диапазоны значений, выявить очевидные проблемы.
- **[code]** `.describe()` на числовых колонках. - **[code]** `.describe()` на числовых колонках.
- **[code]** Поиск аномалий: отрицательные `fare_amount`, нулевая дистанция с ненулевой суммой, `total_amount > $1000`, pickup за пределами ожидаемого диапазона дат (определяется динамически из имён файлов, а не хардкодится). - **[code]** Поиск аномалий: отрицательные `fare_amount`, нулевая дистанция с ненулевой суммой, `total_amount > $1000`, pickup за пределами ожидаемого диапазона дат (определяется динамически из имён файлов, а не хардкодится).
- **[md]** Аномалии в реальных данных — норма, а не баг загрузки. Задача raw — сохранить как есть. - **[md]** Аномалии в реальных данных — норма, а не баг загрузки. Задача raw — сохранить как есть.
### Секция 6: Самостоятельное задание ### Секция 8: Самостоятельное задание
- **[md]** Инструкция: (1) прочитать `taxi_zone_lookup.csv` из raw-зоны через Spark, (2) вывести схему + 10 строк, (3) посчитать количество зон по borough, (4) найти топ-5 pickup locations и join с зонами, (5) найти ещё одну аномалию или интересный паттерн в данных. - **[md]** Инструкция: (1) прочитать `taxi_zone_lookup.csv` из raw-зоны через Spark, (2) вывести схему + 10 строк, (3) посчитать количество зон по borough, (4) найти топ-5 pickup locations и join с зонами, (5) найти ещё одну аномалию или интересный паттерн в данных.
- **[code]** 4 пустых ячейки с `# Ваш код: ...`. - **[code]** 4 пустых ячейки с `# Ваш код: ...`.
### Секция 7: Почему raw-слой важен ### Секция 9: Почему raw-слой важен
- **[md]** Мини-лекция (3-5 абзацев): raw = неизменяемый архив; параллель с stg/ods в Greenplum; почему не скачиваем в ноутбуке; почему не чиним на месте (можно перестроить bronze/silver заново); в Lakehouse raw-файлы лежат в storage отдельно от compute — Spark читает их напрямую, а для Trino потребуется оформить данные в таблицу. - **[md]** Мини-лекция (3-5 абзацев): raw = неизменяемый архив; параллель с stg/ods в Greenplum; почему не скачиваем в ноутбуке; почему не чиним на месте (можно перестроить bronze/silver заново); в Lakehouse raw-файлы лежат в storage отдельно от compute — Spark читает их напрямую, а для Trino потребуется оформить данные в таблицу.
### Секция 8: Checkpoint ### Секция 10: Checkpoint
- **[md]** Вопросы для самопроверки: (1) показать файлы в raw-зоне MinIO, (2) какая схема у Yellow Taxi — сколько колонок, (3) какие колонки содержат null — это баг или норма, (4) зачем raw-слой, если данные потом трансформируются, (5) почему не скачиваем из ноутбука, (6) что произойдёт, если NYC TLC изменит формат файла. - **[md]** Вопросы об источниках настроек, месте выполнения заданий и новом ядре; затем самопроверка по данным: (1) показать файлы в raw-зоне MinIO, (2) какая схема у Yellow Taxi — сколько колонок, (3) какие колонки содержат null — это баг или норма, (4) зачем raw-слой, если данные потом трансформируются, (5) почему не скачиваем из ноутбука, (6) что произойдёт, если NYC TLC изменит формат файла.
### Секция 9: Завершение ### Секция 11: Завершение
- **[md]** «Мы намеренно не удаляем raw-данные из MinIO. В Модуле 4 мы будем использовать их для создания bronze-слоя.» - **[md]** «Мы намеренно не удаляем raw-данные из MinIO. В Модуле 4 мы будем использовать их для создания bronze-слоя.»
- **[code]** `spark.stop()`. - **[code]** `spark.stop()`.
### Дизайн-решения по ноутбуку ### Дизайн-решения по ноутбуку
- **Helper-код:** весь код inline в ноутбуке. Отдельный модуль в `src/` не создаётся — студент должен видеть каждый шаг явно (`boto3.upload_file()`, `spark.read.parquet()`, `.printSchema()`). - **Helper-код:** весь код inline в ноутбуке. Отдельный модуль в `src/` не создаётся — студент должен видеть каждый шаг явно (`boto3.upload_file()`, `spark.read.parquet()`, `.printSchema()`).
- **Cleanup:** ноутбук НЕ удаляет raw-данные из MinIO — они нужны Модулю 4 (bronze). Spark-сессия останавливается. Явное пояснение в конце ноутбука. - **Cleanup:** практика настройки удаляет только свою временную таблицу. Ноутбук НЕ удаляет raw-данные из MinIO — они нужны Модулю 4 (bronze). Spark-сессия останавливается в конце.
## Checkpoint ## Checkpoint
Студент должен уметь: Студент должен уметь:
- создать сессию по заполненному шаблону, подтвердить master `spark://spark-master:7077` и заданное appName;
- прочитать `[0, 1]` из собственной Iceberg-таблицы и подтвердить её удаление;
- объяснить, какие настройки предоставил стенд, где выполняются задания и что повторить после перезапуска ядра;
- показать файлы в raw-зоне MinIO (через MinIO Console или через код); - показать файлы в raw-зоне MinIO (через MinIO Console или через код);
- прочитать raw PARQUET через Spark и вывести несколько строк; - прочитать raw PARQUET через Spark и вывести несколько строк;
- назвать схему Yellow Taxi (количество колонок, основные типы); - назвать схему Yellow Taxi (количество колонок, основные типы);
@@ -148,10 +166,11 @@ volumes:
- `docker-compose.yml` содержит volume mount `./data:/opt/data:ro` для контейнера `jupyter`, и `docker compose config` проходит без ошибок. - `docker-compose.yml` содержит volume mount `./data:/opt/data:ro` для контейнера `jupyter`, и `docker compose config` проходит без ошибок.
- `START_HERE.md` содержит воспроизводимую инструкцию по скачиванию NYC Taxi PARQUET (минимум 3 файла + taxi zone lookup) до запуска Модуля 3. - `START_HERE.md` содержит воспроизводимую инструкцию по скачиванию NYC Taxi PARQUET (минимум 3 файла + taxi zone lookup) до запуска Модуля 3.
- `.gitignore` содержит `/data/` для исключения датасета из git. - `.gitignore` содержит `/data/` для исключения датасета из git.
- `notebooks/03_raw_ingest_and_first_read.ipynb` можно выполнить сверху вниз в поднятом Jupyter-окружении при наличии скачанного data bundle. - `notebooks/03_raw_ingest_and_first_read.ipynb` можно пройти по порядку в поднятом Jupyter-окружении при наличии data bundle, заполнив шаблон и выполнив указанный перезапуск ядра.
- Ноутбук следует методике курса: `объяснение -> демонстрация -> самостоятельное повторение -> checkpoint`. - Ноутбук следует методике курса: `объяснение -> демонстрация -> самостоятельное повторение -> checkpoint`.
- Ноутбук использует паттерны из Модулей 1-2: `boto3` для MinIO, SparkSession с адресом кластера из `spark-defaults.conf`, `setLogLevel("ERROR")`. - Демонстрация берёт master из `spark-defaults.conf`; самостоятельное задание явно задаёт тот же адрес, сохраняя подготовленную конфигурацию каталога. Шаблон и контрольное решение проверяются в новых ядрах на Spark 3.5.1.
- Raw-данные остаются в MinIO после выполнения ноутбука (cleanup отсутствует намеренно). - Проверены повторные ячейки, перезапуск ядра, изменение и восстановление часового пояса, новое имя приложения после остановки контекста.
- Raw-данные остаются в MinIO после выполнения ноутбука; удаляется только временная таблица настройки.
- При отсутствии data bundle — понятный assert с отсылкой к `START_HERE.md`. - При отсутствии data bundle — понятный assert с отсылкой к `START_HERE.md`.
- `docs/stack_reference.md` обновлён: секция JupyterLab содержит `./data` mount. - `docs/stack_reference.md` обновлён: секция JupyterLab содержит `./data` mount.
- `plans/README.md` содержит строку Модуля 3 в оглавлении. - `plans/README.md` содержит строку Модуля 3 в оглавлении.
@@ -159,14 +178,15 @@ volumes:
## Риски ## Риски
- **Размер данных.** 3 месяца Yellow Taxi — примерно 120-150 MB (~40-50 MB на файл). Комфортно для стенда с 10 GB RAM. Extended (12 месяцев, ~600 MB) может замедлить Spark, но не должен вызывать падения. - **Размер данных.** 3 месяца Yellow Taxi — примерно 120-150 MB (~40-50 MB на файл). Комфортно для стенда с 10 GB RAM. Extended (12 месяцев, ~600 MB) может замедлить Spark, но не должен вызывать падения.
- **Студент забыл скачать данные.** Assert в Секции 1 с понятным сообщением и ссылкой на `START_HERE.md`. - **Студент забыл скачать данные.** Assert в Секции 3 с понятным сообщением и ссылкой на `START_HERE.md`.
- **NYC TLC изменит формат/URL.** Ноутбук использует wildcard `yellow_tripdata_*.parquet` без привязки к году. Инструкция в `START_HERE.md` фиксирует конкретные файлы 2024-01..03, но если URL станут недоступны — потребуется обновление инструкции. - **NYC TLC изменит формат/URL.** Ноутбук использует wildcard `yellow_tripdata_*.parquet` без привязки к году. Инструкция в `START_HERE.md` фиксирует конкретные файлы 2024-01..03, но если URL станут недоступны — потребуется обновление инструкции.
- **Директория `data/` не существует.** Docker volume mount создаст пустую директорию, assert в ноутбуке поймает отсутствие файлов. - **Директория `data/` не существует.** Docker volume mount создаст пустую директорию, assert в ноутбуке поймает отсутствие файлов.
- **Пересоздание контейнера.** Добавление volume в `docker-compose.yml` требует пересоздания контейнера `jupyter` (`docker compose up -d` достаточно — Docker пересоздаст только изменённый сервис). После этого bind mount уже работает: новые файлы в `./data` видны сразу без дополнительных перезапусков. Указать явно в инструкции. - **Пересоздание контейнера.** Добавление volume в `docker-compose.yml` требует пересоздания контейнера `jupyter` (`docker compose up -d` достаточно — Docker пересоздаст только изменённый сервис). После этого bind mount уже работает: новые файлы в `./data` видны сразу без дополнительных перезапусков. Указать явно в инструкции.
## Out of Scope ## Out of Scope
- Создание Iceberg-таблиц — отложено до Модуля 4. - Создание рабочих Iceberg-таблиц из raw — в Модуле 4. В Модуле 3 есть только маленькая временная таблица для проверки подключения.
- Подбор памяти, числа executors, shuffle, GC и эксперимент с числом партиций.
- Трансформации и очистка данных — отложено до Модулей 4-5. - Трансформации и очистка данных — отложено до Модулей 4-5.
- Чтение raw-данных через Trino (Trino не умеет читать произвольные файлы без таблицы) — отложено до Модуля 6. - Чтение raw-данных через Trino (Trino не умеет читать произвольные файлы без таблицы) — отложено до Модуля 6.
- Партиционирование, MERGE, streaming, gold-слой. - Партиционирование, MERGE, streaming, gold-слой.