{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Модуль 7. Безопасная работа с таблицами: schema evolution и time travel\n", "\n", "**Цель:** Научиться безопасно изменять Iceberg-таблицы, использовать встроенные механизмы time travel (чтение предыдущих состояний) и rollback (восстановление после ошибок).\n", "\n", "В этом модуле мы сравним подходы классических DWH и Lakehouse к работе с изменениями:\n", "\n", "| | Классический DWH (Greenplum) | Lakehouse (Iceberg) |\n", "|---|---|---|\n", "| Добавить колонку | `ALTER TABLE ADD COLUMN` (мгновенно, NULL) | `ALTER TABLE ADD COLUMNS` (мгновенно, NULL) |\n", "| Переименовать колонку | `ALTER TABLE RENAME COLUMN` | `ALTER TABLE RENAME COLUMN` |\n", "| Получить данные на момент времени в прошлом | Чтение из восстановленного бэкапа (pg_dump / PITR) | `SELECT * FROM TIMESTAMP AS OF ''` |\n", "| Посмотреть историю изменений таблицы | Нет встроенного механизма | `SELECT * FROM table.snapshots` |\n", "| Восстановить таблицу после ошибочной записи | Нужен бэкап или ручной откат | `CALL lakehouse.system.rollback_to_snapshot(table => '.
', snapshot_id => );` |\n", "\n", "Time travel читает прошлое состояние, не меняя текущую таблицу. Rollback переключает текущее состояние таблицы на выбранный snapshot-предок.\n", "\n", "**Ключевая идея:** Iceberg хранит историю изменений данных, пока snapshot-ы не удалены при обслуживании. Schema evolution (изменение структуры) и time travel (путешествие во времени) — встроенные инструменты формата.\n", "\n", "Как устроен этот ноутбук:\n", "- Schema evolution демонстрируем на нашей рабочей таблице `silver` (это безопасная операция).\n", "- Time travel и восстановление после ошибки — на отдельной демо-таблице (чтобы не рисковать данными для Модуля 8).\n", "- Trino используется для проверки эффекта на downstream (системы, читающие данные после нас).\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 1: Spark-сессия, Trino-клиент и проверка таблиц\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "from pyspark.sql import SparkSession\n", "from trino.dbapi import connect\n", "import trino\n", "from IPython.display import display\n", "\n", "spark = SparkSession.builder \\\n", " .appName(\"module-07-safe-table-changes\") \\\n", " .getOrCreate()\n", "\n", "spark.sparkContext.setLogLevel(\"ERROR\")\n", "print(\"SparkSession создана.\")\n", "\n", "SILVER_TABLE = \"lakehouse.silver.nyc_taxi_yellow\"\n", "DEMO_TABLE = \"lakehouse.default.taxi_changes_demo\"\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Проверяем, что silver-таблица существует (Модуль 5 должен быть пройден)\n", "assert spark.catalog.tableExists(SILVER_TABLE), f\"Таблица {SILVER_TABLE} не найдена. Пройдите Модуль 5.\"\n", "count = spark.table(SILVER_TABLE).count()\n", "assert count > 0, f\"Таблица {SILVER_TABLE} пуста.\"\n", "print(f\"Таблица {SILVER_TABLE} готова к работе. Строк: {count}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def trino_query(sql: str) -> pd.DataFrame:\n", " \"\"\"Выполняет SQL-запрос в Trino и возвращает результат как Pandas DataFrame.\"\"\"\n", " conn = connect(\n", " host=\"trino\",\n", " port=8080,\n", " user=\"jupyter\",\n", " catalog=\"lakehouse\"\n", " )\n", " cur = conn.cursor()\n", " try:\n", " cur.execute(sql)\n", " rows = cur.fetchall()\n", " columns = [desc[0] for desc in cur.description]\n", " return pd.DataFrame(rows, columns=columns)\n", " except trino.exceptions.ProgrammingError as e:\n", " if \"No nodes available to run query\" in str(e):\n", " print(\"Ошибка: Trino ещё не готов. Подождите пару минут после старта контейнеров.\")\n", " raise\n", " # Запросы типа CREATE/DROP не возвращают строк\n", " return pd.DataFrame([{\"status\": \"Success\"}])\n", " finally:\n", " conn.close()\n", "\n", "print(\"Trino helper функция готова.\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "*Если ячейки выше упали с `ImportError: trino` — нужно пересобрать образ Jupyter (в терминале: `docker compose build jupyter && docker compose up -d`).*\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Silver на месте, Trino готов. В этом модуле мы будем изменять таблицу и наблюдать последствия из обоих движков.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 2: Snapshot-ы — встроенная история изменений\n", "\n", "В Модуле 4 мы впервые увидели snapshot-ы bronze-таблицы. Теперь разберёмся глубже. Каждая операция с данными в Iceberg (INSERT, overwrite, delete) автоматически создаёт snapshot — «снимок» состояния таблицы. Snapshot фиксирует, какие data files составляли таблицу в этот момент. Это как коммит в git: можно вернуться к любому предыдущему состоянию.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Посмотрим snapshot-историю silver-таблицы\n", "spark.sql(f\"SELECT snapshot_id, committed_at, operation, summary FROM {SILVER_TABLE}.snapshots\").show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Дополнительная история с указанием предков (родительских snapshot-ов)\n", "spark.sql(f\"SELECT * FROM {SILVER_TABLE}.history\").show(truncate=False)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Поля таблицы snapshots:\n", "- `committed_at` — когда произошла операция.\n", "- `operation` — тип операции (append, overwrite).\n", "- `summary` — краткая статистика (added-data-files, total-records и т.д.).\n", "\n", "После первого запуска Модуля 5 у silver будет 1 snapshot от CTAS. Каждый повторный `CREATE OR REPLACE` добавит ещё один snapshot: прежние состояния сохранятся, хотя перестанут быть предками текущего.\n", "В Greenplum такой истории нет: после `INSERT` предыдущее состояние таблицы недоступно без бэкапа.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 3: Schema evolution — безопасное добавление колонки\n", "\n", "Добавляем колонку `loaded_at` к silver-таблице. В PostgreSQL `ALTER TABLE ADD COLUMN` — обычная операция. В Iceberg — аналогично, но с важным свойством: Iceberg хранит историю схем.\n", "\n", "`ALTER TABLE ADD COLUMNS` **не создаёт новый snapshot данных**: data files не перезаписываются, при чтении существующих строк новая колонка возвращает NULL.\n", "\n", "Изменение схемы записывается в новый файл `*.metadata.json` в MinIO. В списке `schemas` появляется новая версия схемы со своим `schema-id`, а `current-schema-id` указывает на неё как на текущую. JDBC-каталог в PostgreSQL переключает `metadata_location` на новый файл. Так схема получает новую версию, хотя текущий snapshot данных остаётся прежним.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "try:\n", " spark.sql(f\"ALTER TABLE {SILVER_TABLE} ADD COLUMNS (loaded_at TIMESTAMP)\")\n", " print(\"Колонка loaded_at добавлена.\")\n", "except Exception as e:\n", " if \"already exists\" in str(e).lower():\n", " print(\"Колонка loaded_at уже существует (повторный запуск ноутбука). Продолжаем.\")\n", " else:\n", " raise\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Проверим схему из Spark\n", "spark.table(SILVER_TABLE).printSchema()\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Посмотрим данные (должен быть NULL)\n", "spark.sql(f\"SELECT VendorID, loaded_at FROM {SILVER_TABLE} LIMIT 5\").show()\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Колонка добавлена. Данные не изменились — Iceberg не перезаписывал data files. \n", "В рабочем сценарии `loaded_at` заполнялся бы при следующих INSERT-ах: `INSERT INTO ... SELECT ..., current_timestamp() AS loaded_at FROM ...`. Существующие строки остаются с NULL — это нормальная практика эволюции схемы. Заполнение существующих строк (UPDATE) — отдельная тема, требующая перезаписи файлов.\n", "\n", "Примечание: если вы повторно запустите Модуль 5 после Модуля 7, `CREATE OR REPLACE TABLE` заменит текущую схему silver, поэтому колонка `loaded_at` исчезнет из текущего состояния. Предыдущие snapshot-ы при этом сохранятся для time travel, но перестанут быть предками текущего snapshot-а.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 4: Downstream-эффект — Trino видит изменение\n", "\n", "В Модуле 6 мы убедились: Spark и Trino читают одну таблицу через общий каталог. Schema evolution — ещё одно следствие этого дизайна: изменение схемы в Spark мгновенно видно в Trino. Не нужно «синхронизировать» или «обновлять» что-то на стороне Trino.\n", "\n", "*В DBeaver: `DESCRIBE lakehouse.silver.nyc_taxi_yellow;`*\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Проверим схему из Trino\n", "display(trino_query(f\"DESCRIBE {SILVER_TABLE}\"))\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Прочитаем данные из Trino\n", "display(trino_query(f\"SELECT vendorid, loaded_at FROM {SILVER_TABLE} LIMIT 5\"))\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Trino увидел новую колонку. Spark записал новый metadata-файл в MinIO и обновил указатель на него в PostgreSQL (JDBC catalog). Trino прочитал схему из нового файла. Decoupled compute + общий каталог в действии.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 5: Обзор операций schema evolution\n", "\n", "`ADD COLUMN` — самая безопасная операция. Но Iceberg поддерживает и другие:\n", "\n", "| Операция | Spark SQL | Безопасность | Влияние на downstream |\n", "|---|---|---|---|\n", "| Добавить колонку | `ALTER TABLE ADD COLUMNS (col type)` | Безопасно | Новая колонка с NULL, существующие запросы не ломаются |\n", "| Переименовать колонку | `ALTER TABLE RENAME COLUMN old TO new` | Осторожно | Запросы с `SELECT old_name` перестают работать |\n", "| Расширить тип | `ALTER TABLE ALTER COLUMN col TYPE bigint` | Безопасно | INT -> BIGINT: без потери данных |\n", "| Сузить тип | — | Опасно | BIGINT -> INT: возможна потеря данных. Iceberg не поддерживает |\n", "| Удалить колонку | `ALTER TABLE DROP COLUMN col` | Опасно | Запросы с `SELECT col` перестают работать |\n", "\n", "Далее в модуле мы продемонстрируем `RENAME COLUMN` на демо-таблице и покажем, как это ломает downstream-запросы.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 6: Подготовка демо-таблицы с несколькими snapshot-ами\n", "\n", "Для экспериментов с time travel и восстановлением (rollback) создаём отдельную демо-таблицу. Silver не трогаем — он нужен для финального Модуля 8. \n", "\n", "Мы создадим таблицу с небольшим подмножеством silver (1000 строк) и нарастим snapshot-историю через дополнительные `INSERT INTO`.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Демо-таблица одноразовая: удаляем её вместе с историей,\n", "# чтобы при повторном запуске снова получить ровно один snapshot.\n", "spark.sql(f\"DROP TABLE IF EXISTS {DEMO_TABLE} PURGE\")\n", "\n", "# Создание демо-таблицы\n", "spark.sql(f\"\"\"\n", " CREATE TABLE {DEMO_TABLE}\n", " USING iceberg AS\n", " SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,\n", " passenger_count, trip_distance, fare_amount, total_amount,\n", " pickup_borough, pickup_zone\n", " FROM {SILVER_TABLE}\n", " LIMIT 1000\n", "\"\"\")\n", "\n", "count_1 = spark.table(DEMO_TABLE).count()\n", "assert count_1 == 1000, f\"Ожидалось 1000 строк, получено {count_1}\"\n", "print(f\"Таблица создана. Строк: {count_1}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Сохраним ID первого snapshot-а\n", "df_snapshots = spark.sql(f\"SELECT snapshot_id, committed_at FROM {DEMO_TABLE}.snapshots ORDER BY committed_at ASC\").collect()\n", "assert len(df_snapshots) == 1, f\"Ожидался 1 snapshot, получено {len(df_snapshots)}\"\n", "snapshot_1 = df_snapshots[0]['snapshot_id']\n", "committed_at_1 = df_snapshots[0]['committed_at']\n", "\n", "print(f\"Snapshot 1 ID: {snapshot_1} (создан: {committed_at_1})\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Добавим ещё 500 строк из Манхэттена (создаст второй snapshot)\n", "spark.sql(f\"\"\"\n", " INSERT INTO {DEMO_TABLE}\n", " SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,\n", " passenger_count, trip_distance, fare_amount, total_amount,\n", " pickup_borough, pickup_zone\n", " FROM {SILVER_TABLE}\n", " WHERE pickup_borough = 'Manhattan'\n", " LIMIT 500\n", "\"\"\")\n", "\n", "count_2 = spark.table(DEMO_TABLE).count()\n", "print(f\"Добавлено 500 строк. Всего строк: {count_2}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Сохраним ID второго snapshot-а\n", "df_snapshots = spark.sql(f\"SELECT snapshot_id, committed_at FROM {DEMO_TABLE}.snapshots ORDER BY committed_at ASC\").collect()\n", "snapshot_2 = df_snapshots[1]['snapshot_id']\n", "committed_at_2 = df_snapshots[1]['committed_at']\n", "\n", "print(f\"Snapshot 2 ID: {snapshot_2} (создан: {committed_at_2})\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Теперь у таблицы два snapshot-а: начальная загрузка (1000 строк) и добавление (500 строк). Это как два коммита в git. Каждый фиксирует конкретное состояние таблицы.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 7: Time travel — чтение предыдущих состояний\n", "\n", "Центральная возможность Iceberg: можно прочитать таблицу в том состоянии, в каком она была на момент любого snapshot-а. Это time travel. В Greenplum для этого пришлось бы восстанавливать бэкап целой базы или настраивать сложный PITR.\n", "\n", "#### 7a: Time travel через Spark\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Чтение первого snapshot-а (начальная загрузка)\n", "print(f\"Читаем таблицу по состоянию Snapshot 1 ({snapshot_1}):\")\n", "\n", "spark.sql(f\"\"\"\n", " SELECT count(*) AS row_count\n", " FROM {DEMO_TABLE}\n", " VERSION AS OF {snapshot_1}\n", "\"\"\").show()\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Чтение текущего состояния для сравнения\n", "print(\"Читаем текущее состояние таблицы:\")\n", "\n", "spark.sql(f\"\"\"\n", " SELECT count(*) AS row_count\n", " FROM {DEMO_TABLE}\n", "\"\"\").show()\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Данные из первого snapshot-а не потеряны — оба состояния доступны одновременно. Iceberg хранит все data files; snapshot определяет, какие из них составляют таблицу в конкретный момент.\n", "\n", "#### 7b: Time travel через Trino\n", "\n", "Trino тоже умеет time travel. Тот же snapshot, та же таблица, тот же каталог.\n", "Обратите внимание на синтаксис: в Spark — `VERSION AS OF`, в Trino — `FOR VERSION AS OF`.\n", "\n", "*В DBeaver: `SELECT count(*) FROM lakehouse.default.taxi_changes_demo FOR VERSION AS OF ;`*\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Time travel через Trino к первому snapshot-у\n", "print(f\"Trino читает Snapshot 1 ({snapshot_1}):\")\n", "\n", "display(trino_query(f\"\"\"\n", " SELECT count(*) AS row_count\n", " FROM {DEMO_TABLE}\n", " FOR VERSION AS OF {snapshot_1}\n", "\"\"\"))\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "#### 7c: Time travel по времени\n", "\n", "Альтернативный способ — по timestamp. `TIMESTAMP AS OF` читает snapshot, который был текущим в указанный момент времени. Чтобы прочитать конкретный snapshot по его ID, используй `VERSION AS OF`. Оба запроса оставляют текущее состояние таблицы без изменений.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Time travel по времени (состояние на момент создания Snapshot 1)\n", "spark.sql(f\"\"\"\n", " SELECT count(*) AS row_count\n", " FROM {DEMO_TABLE}\n", " TIMESTAMP AS OF '{committed_at_1}'\n", "\"\"\").show()\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 8: Намеренная ошибка и восстановление (Rollback)\n", "\n", "Ключевой урок модуля. Мы намеренно внесём «плохие» данные, затем восстановим таблицу через rollback. Ошибки в Iceberg — не катастрофа, если понимаешь snapshot-историю.\n", "\n", "#### 8a: Вносим ошибку\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# INSERT «плохих» данных (200 строк с отрицательным fare_amount)\n", "spark.sql(f\"\"\"\n", " INSERT INTO {DEMO_TABLE}\n", " SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,\n", " passenger_count, trip_distance,\n", " -99.99 AS fare_amount,\n", " -99.99 AS total_amount,\n", " 'ERROR' AS pickup_borough,\n", " 'BAD_DATA' AS pickup_zone\n", " FROM {SILVER_TABLE}\n", " LIMIT 200\n", "\"\"\")\n", "\n", "# Убедимся, что таблица \"испорчена\"\n", "spark.sql(f\"\"\"\n", " SELECT count(*) as bad_rows \n", " FROM {DEMO_TABLE} \n", " WHERE fare_amount < 0\n", "\"\"\").show()\n", "\n", "count_3 = spark.table(DEMO_TABLE).count()\n", "print(f\"Всего строк сейчас: {count_3}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Посмотрим на 3-й snapshot (нашу ошибку)\n", "spark.sql(f\"SELECT snapshot_id, parent_id, operation FROM {DEMO_TABLE}.snapshots\").show(truncate=False)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "В Greenplum: 200 ошибочных строк в production-таблице -> звонок DBA, восстановление из бэкапа (если он есть и свежий), простой сервиса, или опасные ручные `DELETE`/`UPDATE` без возможности отката.\n", "В Iceberg: rollback к предыдущему snapshot-у.\n", "\n", "#### 8b: Восстановление через rollback\n", "Мы откатимся к snapshot 2, который мы сохранили в переменной `snapshot_2`.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Выполняем rollback\n", "spark.sql(f\"\"\"\n", " CALL lakehouse.system.rollback_to_snapshot(\n", " table => 'default.taxi_changes_demo',\n", " snapshot_id => {snapshot_2}\n", " )\n", "\"\"\")\n", "print(f\"Rollback к snapshot {snapshot_2} завершен.\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Проверка восстановления\n", "bad_rows = spark.sql(f\"SELECT count(*) as bad_rows FROM {DEMO_TABLE} WHERE fare_amount < 0\").collect()[0]['bad_rows']\n", "current_count = spark.table(DEMO_TABLE).count()\n", "\n", "print(f\"Плохих строк (fare_amount < 0): {bad_rows}\")\n", "print(f\"Всего строк: {current_count}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Таблица восстановлена!\n", "\n", "#### 8c: Что произошло со snapshot-историей?\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print(\"Сохранённые snapshot-ы:\")\n", "spark.sql(f\"SELECT snapshot_id, parent_id, operation FROM {DEMO_TABLE}.snapshots\").show(truncate=False)\n", "\n", "print(\"История переключений текущего состояния:\")\n", "spark.sql(\n", " f\"SELECT made_current_at, snapshot_id, parent_id, is_current_ancestor \"\n", " f\"FROM {DEMO_TABLE}.history ORDER BY made_current_at\"\n", ").show(truncate=False)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "В `.snapshots` по-прежнему три записи: rollback **не создал четвёртый snapshot**. Он переключил текущее состояние таблицы на уже существующий `snapshot_2`.\n", "\n", "В `.history` появилась ещё одна запись о переключении текущего состояния. Она снова указывает на `snapshot_2`; snapshot с «плохими» строками сохранился, но больше не является предком текущего состояния (`is_current_ancestor = false`).\n", "\n", "| Свойство | `rollback_to_snapshot` | `CREATE OR REPLACE TABLE ... AS SELECT` |\n", "|---|---|---|\n", "| Создаёт новый snapshot | Нет | Да |\n", "| Что становится текущим | Существующий snapshot-предок | Новый snapshot с результатом `SELECT` |\n", "| Что происходит со старыми snapshot-ами | Сохраняются | Сохраняются для time travel |\n", "| Цепочка предков | Возвращаемся к предку | Новый snapshot не имеет `parent_id`; прежние snapshot-ы перестают быть предками текущего |\n", "\n", "При `CREATE OR REPLACE` отдельная именованная ветка автоматически не создаётся: текущей остаётся ветка `main`. Старые data files сохраняются до `expire_snapshots` — эту процедуру разберём в Модуле 8.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 9: RENAME COLUMN и влияние на downstream\n", "\n", "Покажем, что `RENAME COLUMN` — безопасная операция для данных (они не перезаписываются), но опасная для downstream-запросов (они могут сломаться).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Переименуем колонку в демо-таблице\n", "spark.sql(f\"ALTER TABLE {DEMO_TABLE} RENAME COLUMN pickup_borough TO borough\")\n", "print(\"Колонка переименована в Spark.\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Посмотрим схему из Spark - колонка теперь borough\n", "spark.table(DEMO_TABLE).printSchema()\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Запрос с новым именем borough через Trino работает\n", "display(trino_query(f\"SELECT borough FROM {DEMO_TABLE} LIMIT 3\"))\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# А запрос со старым именем pickup_borough упадет с ошибкой\n", "try:\n", " display(trino_query(f\"SELECT pickup_borough FROM {DEMO_TABLE} LIMIT 3\"))\n", "except Exception as e:\n", " print(f\"Ошибка в Trino: {e}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Данные не потеряны — колонка переименована в метаданных. Но любой downstream-запрос, использовавший старое имя `pickup_borough`, сломается. В рабочем окружении перед `RENAME` нужно проверить, кто и как читает эту колонку.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 10: Типичные ошибки новичка\n", "\n", "**Ошибка 1: Слепой `CREATE OR REPLACE`.**\n", "В Модулях 4 и 5 мы использовали `CREATE OR REPLACE TABLE ... AS SELECT` для воспроизводимой полной загрузки bronze и silver. В `SparkCatalog` эта операция атомарно создаёт новый snapshot и сохраняет прежние snapshot-ы для time travel. Но она заменяет текущее содержимое, схему и партиционирование таблицы. Новый snapshot не имеет `parent_id`, поэтому прежние snapshot-ы перестают быть предками текущего состояния. Слепая замена может неожиданно убрать колонку, изменить раскладку данных или усложнить обычный rollback к прежней цепочке. Если нужно перезаписать данные, сохранив текущую схему и партиционирование, используй `INSERT OVERWRITE`.\n", "\n", "**Ошибка 2: Изменение схемы без проверки downstream.**\n", "Мы только что видели, как `RENAME COLUMN` ломает запросы со старым именем. `ALTER COLUMN TYPE` (расширение INT -> BIGINT) безопасен для данных, но downstream-системы могут интерпретировать новый тип по-другому и сломаться на своей стороне. Перед любым изменением схемы нужно понимать, кто читает эту таблицу.\n", "\n", "**Ошибка 3: Отсутствие проверки snapshot-истории перед деструктивной операцией.**\n", "Перед любой сложной операцией, массово изменяющей данные (UPDATE/DELETE), полезно посмотреть текущее состояние: `SELECT * FROM table.snapshots`. Это занимает секунды и даёт понимание, какой snapshot станет точкой отката в случае проблемы. Привычка: посмотрел snapshot-ы -> выполнил операцию -> проверил результат -> убедился, что новый snapshot создан корректно.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 11: Самостоятельное задание\n", "\n", "Подготовим демо-таблицу с чистого листа. После демонстраций выше её схема изменена, а история содержит учебные эксперименты. Поскольку `CREATE OR REPLACE` сохраняет старые snapshot-ы, для одноразовой учебной таблицы явно используем `DROP TABLE ... PURGE`, а затем новый `CREATE TABLE`. Так самостоятельная работа снова начнётся ровно с одного snapshot-а.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Эта таблица одноразовая: PURGE удаляет её данные и snapshot-историю.\n", "spark.sql(f\"DROP TABLE IF EXISTS {DEMO_TABLE} PURGE\")\n", "\n", "# Создаём новую таблицу с одним начальным snapshot-ом.\n", "spark.sql(f\"\"\"\n", " CREATE TABLE {DEMO_TABLE}\n", " USING iceberg AS\n", " SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,\n", " passenger_count, trip_distance, fare_amount, total_amount,\n", " pickup_borough, pickup_zone\n", " FROM {SILVER_TABLE}\n", " LIMIT 1000\n", "\"\"\")\n", "practice_snapshot_count = spark.sql(f\"SELECT * FROM {DEMO_TABLE}.snapshots\").count()\n", "assert practice_snapshot_count == 1, f\"Ожидался 1 snapshot, получено {practice_snapshot_count}\"\n", "print(f\"Демо-таблица создана начисто: {spark.table(DEMO_TABLE).count()} строк, 1 snapshot.\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Выполни 5 задач на демо-таблице `lakehouse.default.taxi_changes_demo`.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Задача 1 (schema evolution).** Добавь колонку `quality_flag STRING` к демо-таблице через `ALTER TABLE ADD COLUMNS`. Проверь из Spark и из Trino, что колонка появилась и все значения NULL.\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "*DBeaver: `SELECT * FROM lakehouse.default.taxi_changes_demo LIMIT 10;`*\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: ALTER TABLE ADD COLUMNS (quality_flag)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: проверка из Spark и Trino\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Задача 2 (создание snapshot).** Вставь (INSERT INTO) 300 строк из silver (с условием `pickup_borough = 'Brooklyn'`) в демо-таблицу. Проверь, что появился новый snapshot (всего их станет 2).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: INSERT 300 строк из Brooklyn\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Задача 3 (time travel).** Посмотри snapshot-историю демо-таблицы. Прочитай таблицу в состоянии ДО добавления 300 строк (т.е. к snapshot 1) через `VERSION AS OF`. Сравни количество строк.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: snapshot-история\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: time travel — VERSION AS OF\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Задача 4 (recovery).** Сделай INSERT 100 строк с `fare_amount = -1` и `pickup_zone = 'STUDENT_ERROR'` в демо-таблицу. Убедись, что «плохие» строки появились. Затем выполни `rollback_to_snapshot` к snapshot-у ДО этого INSERT-а (т.е. к snapshot 2). Проверь, что таблица восстановлена.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: INSERT 100 «плохих» строк\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: rollback_to_snapshot\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: проверка восстановления\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Задача 5 (ответ).** Сравни `rollback_to_snapshot` и `CREATE OR REPLACE TABLE ... AS SELECT * FROM table VERSION AS OF `. Какая операция создаёт новый snapshot? Что происходит с текущим состоянием, старыми snapshot-ами и цепочкой `parent_id`?\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Ваш ответ на Задачу 5: \n", "...\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "*(Дополнительно, по желанию)*: Посмотри snapshot-историю silver-таблицы. Сколько snapshot-ов у неё? Какие операции их создали?\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Ваш код: snapshot-история silver\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 12: Checkpoint\n", "\n", "Ответь для себя на вопросы:\n", "1. Что такое snapshot в Iceberg и когда он создаётся?\n", "2. Создаёт ли `ALTER TABLE ADD COLUMNS` новый data snapshot? Почему?\n", "3. Чем `ADD COLUMN` отличается от `RENAME COLUMN` по влиянию на downstream?\n", "4. Как прочитать предыдущее состояние таблицы? (назови два способа)\n", "5. Что делает `rollback_to_snapshot`? Теряется ли при этом snapshot-история?\n", "6. Какие риски остаются у `CREATE OR REPLACE`, хотя старые snapshot-ы сохраняются?\n", "7. Какой аналог time travel в классических СУБД (PostgreSQL/Greenplum) и почему он сложнее в использовании?\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Секция 13: Завершение\n", "\n", "Удаляем демо-таблицу — она больше не нужна. Silver остаётся для Модуля 8 (колонка `loaded_at` — демонстрационная; Модуль 8 не зависит от неё).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "spark.sql(f\"DROP TABLE IF EXISTS {DEMO_TABLE} PURGE\")\n", "print(\"Демо-таблица и её учебные данные удалены.\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Итог модуля:**\n", "- Schema evolution (`ADD COLUMN`) — безопасная metadata-only операция, которая мгновенно видна из всех движков через общий каталог.\n", "- Snapshot-ы — автоматическая история изменений данных, встроенная в Iceberg.\n", "- Time travel — чтение предыдущих состояний без бэкапов.\n", "- Rollback — переключение текущего состояния на существующий snapshot-предок без создания нового snapshot-а.\n", "\n", "В финальном Модуле 8 мы изучим базовое обслуживание таблиц: compaction (решение проблемы множества мелких файлов) и `expire_snapshots` (очистка старых файлов данных, оставшихся после time travel и rollback, которые мы научились применять в этом модуле).\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "spark.stop()\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 4 }