ddadmin
91a5c85469
- Зачем: модули 2–8 в новом процессе выбирали local[*] вместо Spark-кластера. - Что: master перенесён в общий spark-defaults.conf, smoke требует точный адрес кластера, документация и планы синхронизированы. - Проверка: фактическая инициализация всех восьми ноутбуков, повторный getOrCreate, распределённые вычисления, Iceberg/MinIO write-read-cleanup и docker compose config прошли.
301 lines
13 KiB
Plaintext
301 lines
13 KiB
Plaintext
{
|
||
"cells": [
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"# Модуль 2. Ментальная модель Lakehouse: storage, catalog, compute\n",
|
||
"\n",
|
||
"В этом модуле мы на практике разберем, чем Lakehouse отличается от классических реляционных СУБД (PostgreSQL, Greenplum).\n",
|
||
"В классической СУБД данные, метаданные и вычислитель жестко связаны внутри одной системы. В Lakehouse эти слои физически разделены:\n",
|
||
"- **Storage (Хранилище)**: где лежат физические файлы (у нас это MinIO / S3).\n",
|
||
"- **Catalog (Каталог)**: где хранится информация о таблицах и их расположении (у нас это PostgreSQL).\n",
|
||
"- **Compute (Вычислитель)**: кто обрабатывает данные (у нас это Spark, а позже добавится Trino).\n",
|
||
"\n",
|
||
"| | PostgreSQL / Greenplum | Lakehouse |\n",
|
||
"|---|---|---|\n",
|
||
"| **Данные** | Внутри СУБД (pg_data) | Файлы в объектном хранилище (MinIO) |\n",
|
||
"| **Метаданные** | Системные каталоги (pg_catalog) | Внешний каталог (PostgreSQL JDBC) |\n",
|
||
"| **Вычислитель** | Тот же процесс СУБД | Независимые движки (Spark, Trino) |"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"# Адрес Spark-кластера и настройки Iceberg берутся из spark-defaults.conf.\n",
|
||
"# Поэтому в ноутбуке достаточно задать имя приложения.\n",
|
||
"from pyspark.sql import SparkSession\n",
|
||
"\n",
|
||
"spark = SparkSession.builder \\\n",
|
||
" .appName(\"Module-02-Mental-Model\") \\\n",
|
||
" .getOrCreate()\n",
|
||
"\n",
|
||
"# Spark выводит много INFO/WARN логов — убираем их, чтобы не засорять вывод ноутбука\n",
|
||
"spark.sparkContext.setLogLevel(\"ERROR\")\n",
|
||
"print(\"Spark сессия готова!\")"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 1. Запись таблицы (Compute)\n",
|
||
"Мы используем Spark (вычислитель) для создания таблицы и записи данных."
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"# Создаем namespace (логическую базу данных)\n",
|
||
"spark.sql(\"CREATE NAMESPACE IF NOT EXISTS lakehouse.module_02\")\n",
|
||
"\n",
|
||
"# Создаем или перезаписываем таблицу и вставляем пару строк\n",
|
||
"spark.sql(\"\"\"\n",
|
||
"CREATE OR REPLACE TABLE lakehouse.module_02.demo_table (\n",
|
||
" id int,\n",
|
||
" name string,\n",
|
||
" created_at timestamp\n",
|
||
")\n",
|
||
"USING iceberg\n",
|
||
"\"\"\")\n",
|
||
"\n",
|
||
"spark.sql(\"\"\"\n",
|
||
"INSERT INTO lakehouse.module_02.demo_table \n",
|
||
"VALUES \n",
|
||
" (1, 'Alice', current_timestamp()),\n",
|
||
" (2, 'Bob', current_timestamp())\n",
|
||
"\"\"\")\n",
|
||
"\n",
|
||
"# Прочитаем данные через Spark, чтобы убедиться, что они записались\n",
|
||
"spark.sql(\"SELECT * FROM lakehouse.module_02.demo_table\").show()"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"## 2. Где живут метаданные? (Catalog)\n",
|
||
"Таблица в Lakehouse — это не просто файлы. Чтобы Spark (или Trino) знал, где лежат данные и какая у них схема, используется Catalog. В нашем стенде это PostgreSQL.\n",
|
||
"Посмотрим, какая запись появилась в базе данных каталога."
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"import psycopg2\n",
|
||
"import pandas as pd\n",
|
||
"import warnings\n",
|
||
"\n",
|
||
"# Подключаемся к PostgreSQL, который хранит метаданные Iceberg (наш Catalog)\n",
|
||
"with psycopg2.connect(\n",
|
||
" host=\"postgres-iceberg\",\n",
|
||
" database=\"iceberg\",\n",
|
||
" user=\"iceberg\",\n",
|
||
" password=\"iceberg\"\n",
|
||
") as conn:\n",
|
||
" # Запрашиваем информацию о нашей таблице\n",
|
||
" query = \"\"\"\n",
|
||
" SELECT table_namespace, table_name, metadata_location, previous_metadata_location \n",
|
||
" FROM iceberg_tables \n",
|
||
" WHERE table_namespace = 'module_02'\n",
|
||
" \"\"\"\n",
|
||
"\n",
|
||
" # Используем catch_warnings локально, чтобы скрыть UserWarning от pandas при чтении из DB-API\n",
|
||
" with warnings.catch_warnings():\n",
|
||
" warnings.simplefilter('ignore', UserWarning)\n",
|
||
" df_catalog = pd.read_sql(query, conn)\n",
|
||
"\n",
|
||
"# Посмотрим, куда указывает catalog (обратите внимание на metadata_location)\n",
|
||
"df_catalog"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"В поле `metadata_location` хранится путь к актуальному файлу метаданных таблицы в физическом хранилище.\n",
|
||
"\n",
|
||
"## 3. Где живут сами данные? (Storage)\n",
|
||
"В качестве физического хранилища (Storage) мы используем MinIO — S3-совместимое объектное хранилище. Данные и файлы метаданных лежат там.\n",
|
||
"Посмотрим на физические артефакты нашей таблицы."
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"import boto3\n",
|
||
"\n",
|
||
"# Подключаемся к MinIO\n",
|
||
"s3 = boto3.client(\n",
|
||
" 's3',\n",
|
||
" endpoint_url='http://minio:9000',\n",
|
||
" aws_access_key_id='minioadmin',\n",
|
||
" aws_secret_access_key='minioadmin'\n",
|
||
")\n",
|
||
"\n",
|
||
"bucket_name = 'lakehouse'\n",
|
||
"# Путь (prefix) начинается с warehouse, так как это задано в настройках каталога\n",
|
||
"prefix = 'warehouse/module_02/demo_table/'\n",
|
||
"\n",
|
||
"# Получаем список всех файлов (объектов), относящихся к нашей таблице\n",
|
||
"response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)\n",
|
||
"\n",
|
||
"print(\"Файлы таблицы в MinIO:\\n\")\n",
|
||
"contents = response.get('Contents', [])\n",
|
||
"if not contents:\n",
|
||
" print(\"Файлы не найдены. Проверьте prefix или убедитесь, что таблица создана.\")\n",
|
||
"else:\n",
|
||
" for obj in contents:\n",
|
||
" key = obj['Key']\n",
|
||
" size_kb = obj['Size'] / 1024\n",
|
||
" if '/data/' in key:\n",
|
||
" print(f\"[DATA] {key} ({size_kb:.2f} KB)\")\n",
|
||
" elif '/metadata/' in key:\n",
|
||
" print(f\"[META] {key} ({size_kb:.2f} KB)\")\n",
|
||
" else:\n",
|
||
" print(f\"[OTHER] {key} ({size_kb:.2f} KB)\")"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"**Что мы видим?**\n",
|
||
"1. Папка `metadata/` содержит файлы `.json` (снимки состояния таблицы), `.avro` (манифесты, описывающие, какие файлы данных актуальны).\n",
|
||
"2. Папка `data/` содержит сами данные в формате `.parquet`.\n",
|
||
"\n",
|
||
"> **Подумайте:** Что произойдет, если вы удалите файлы из MinIO, но не удалите запись в каталоге? А что если наоборот?"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"---\n",
|
||
"\n",
|
||
"## 4. Самостоятельное задание\n",
|
||
"\n",
|
||
"1. Создайте таблицу `lakehouse.module_02.my_first_table`.\n",
|
||
"2. Запишите в неё несколько произвольных строк.\n",
|
||
"3. Прочитайте таблицу через Spark.\n",
|
||
"4. Выполните запросы к PostgreSQL и MinIO (аналогично примерам выше), чтобы найти `metadata_location` и физические `parquet` файлы вашей новой таблицы."
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"# Ваш код: создание таблицы my_first_table и вставка данных\n",
|
||
"\n"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"# Ваш код: чтение данных через Spark\n",
|
||
"\n"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"# Ваш код: инспекция каталога (PostgreSQL)\n",
|
||
"\n"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"# Ваш код: инспекция хранилища (MinIO)\n",
|
||
"\n"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"---\n",
|
||
"\n",
|
||
"## Checkpoint\n",
|
||
"\n",
|
||
"Убедитесь, что вы можете ответить на следующие вопросы:\n",
|
||
"- Можете ли вы показать конкретные файлы данных и метаданных созданной таблицы в MinIO (через UI MinIO на порту 9001 или через код выше)?\n",
|
||
"- Можете ли вы показать запись о таблице в PostgreSQL и объяснить, на что указывает поле `metadata_location`?\n",
|
||
"- Что произойдёт, если удалить файлы из MinIO, но не трогать каталог (или наоборот)?\n",
|
||
"- Понимаете ли вы теперь разницу между `storage` (MinIO) и `catalog` (PostgreSQL)?"
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "markdown",
|
||
"metadata": {},
|
||
"source": [
|
||
"---\n",
|
||
"\n",
|
||
"## 5. Очистка ресурсов (Cleanup)\n",
|
||
"Выполните ячейку ниже, чтобы удалить созданные таблицы и namespace. Это необходимо, чтобы не оставлять \"мусор\" перед переходом к следующим модулям."
|
||
]
|
||
},
|
||
{
|
||
"cell_type": "code",
|
||
"execution_count": null,
|
||
"metadata": {},
|
||
"outputs": [],
|
||
"source": [
|
||
"# PURGE удаляет одноразовые таблицы и из каталога, и из хранилища\n",
|
||
"spark.sql(\"DROP TABLE IF EXISTS lakehouse.module_02.demo_table PURGE\")\n",
|
||
"spark.sql(\"DROP TABLE IF EXISTS lakehouse.module_02.my_first_table PURGE\")\n",
|
||
"\n",
|
||
"# Удаляем namespace\n",
|
||
"spark.sql(\"DROP NAMESPACE IF EXISTS lakehouse.module_02\")\n",
|
||
"\n",
|
||
"spark.stop()\n",
|
||
"print(\"Очистка завершена, Spark сессия остановлена.\")"
|
||
]
|
||
}
|
||
],
|
||
"metadata": {
|
||
"kernelspec": {
|
||
"display_name": "Python 3",
|
||
"language": "python",
|
||
"name": "python3"
|
||
},
|
||
"language_info": {
|
||
"codemirror_mode": {
|
||
"name": "ipython",
|
||
"version": 3
|
||
},
|
||
"file_extension": ".py",
|
||
"mimetype": "text/x-python",
|
||
"name": "python",
|
||
"nbconvert_exporter": "python",
|
||
"pygments_lexer": "ipython3",
|
||
"version": "3.10.12"
|
||
}
|
||
},
|
||
"nbformat": 4,
|
||
"nbformat_minor": 5
|
||
}
|