From bf3241124cb8a63dea57b99e897b23aa4e99f907 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sun, 20 Sep 2026 16:14:53 -0400 Subject: [PATCH] fix(course): clarify history and fix snapshot expiration Set an explicit expiration cutoff because retain_last alone preserves fresh snapshots under the default five-day age threshold. Add result checks, distinguish time travel from rollback, and explain schema metadata versions. Validated changed cells with Spark 3.5.1 and Iceberg 1.10.0 in an isolated container: 10 to 2 snapshots, 1300 rows preserved, expired time travel rejected, repeat execution in UTC and America/New_York. Notebook validation, Python syntax, local links and diff checks pass. --- docs/glossary.md | 4 +- docs/stack_reference.md | 8 ++- notebooks/07_safe_table_changes.ipynb | 16 +++--- notebooks/08_maintenance_and_final_lab.ipynb | 51 +++++++++++++++----- 4 files changed, 57 insertions(+), 22 deletions(-) diff --git a/docs/glossary.md b/docs/glossary.md index d89dd22..8c19633 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -128,9 +128,9 @@ Avro-файл со списком manifest-ов, составляющих оди ### Vacuum (expire_snapshots) -Удаление старых snapshot-ов и связанных с ними data files из хранилища. Освобождает место, но делает невозможным time travel к удалённым snapshot-ам. Необратимая операция. +Удаление старых snapshot-ов и файлов, которые больше не нужны ни одному сохранённому snapshot-у. Может освободить место, но делает невозможным time travel к удалённым snapshot-ам. Необратимая операция. -**В этом курсе:** `CALL lakehouse.system.expire_snapshots(table => '...', retain_last => N)`. +**В этом курсе:** `CALL lakehouse.system.expire_snapshots(table => '...', older_than => TIMESTAMP '', retain_last => N)`. `older_than` задаёт границу времени, а `retain_last` защищает последние N snapshot-ов в цепочке предков текущего состояния независимо от возраста. Без явной границы при стандартных настройках удаляются только snapshot-ы старше 5 дней; один `retain_last` не сокращает историю до N записей. **Параллель с DWH:** `VACUUM` в PostgreSQL/Greenplum (удаление мёртвых строк). **Где в курсе:** Модуль 8. diff --git a/docs/stack_reference.md b/docs/stack_reference.md index 8c21178..3537a06 100644 --- a/docs/stack_reference.md +++ b/docs/stack_reference.md @@ -221,7 +221,11 @@ SELECT * FROM .files; -- Обслуживание (Модуль 8) CALL lakehouse.system.rewrite_data_files(table => '.
'); -CALL lakehouse.system.expire_snapshots(table => '.
', retain_last => N); +CALL lakehouse.system.expire_snapshots( + table => '.
', + older_than => TIMESTAMP '', + retain_last => N +); -- Rollback (Модуль 7) CALL lakehouse.system.rollback_to_snapshot(table => '.
', snapshot_id => ); @@ -230,6 +234,8 @@ CALL lakehouse.system.rollback_to_snapshot(table => '.
', snaps SELECT * FROM
VERSION AS OF ; ``` +Для `expire_snapshots` подставь в `` границу хранения истории в часовом поясе Spark-сессии. `retain_last` сохраняет минимум N последних snapshot-ов в цепочке предков текущего состояния, а не ограничивает всю историю числом N. Получение границы времени для учебного примера показано в [Модуле 8, Секции 4](../notebooks/08_maintenance_and_final_lab.ipynb). + ### Часто используемые Trino SQL ```sql diff --git a/notebooks/07_safe_table_changes.ipynb b/notebooks/07_safe_table_changes.ipynb index b36afa7..09e73e3 100644 --- a/notebooks/07_safe_table_changes.ipynb +++ b/notebooks/07_safe_table_changes.ipynb @@ -14,11 +14,13 @@ "|---|---|---|\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 ... VERSION AS OF ` |\n", + "| Получить данные на момент времени в прошлом | Чтение из восстановленного бэкапа (pg_dump / PITR) | `SELECT * FROM
TIMESTAMP AS OF ''` |\n", "| Посмотреть историю изменений таблицы | Нет встроенного механизма | `SELECT * FROM table.snapshots` |\n", - "| Последствия ошибочной записи | Нужен бэкап или ручной откат | Rollback к предыдущему snapshot |\n", + "| Восстановить таблицу после ошибочной записи | Нужен бэкап или ручной откат | `CALL lakehouse.system.rollback_to_snapshot(table => '.
', snapshot_id => );` |\n", "\n", - "**Ключевая идея:** Iceberg хранит полную историю изменений данных. Schema evolution (изменение структуры) и time travel (путешествие во времени) — встроенные инструменты формата, а не дополнительная инфраструктура.\n", + "Time travel читает прошлое состояние, не меняя текущую таблицу. Rollback переключает текущее состояние таблицы на выбранный snapshot-предок.\n", + "\n", + "**Ключевая идея:** Iceberg хранит историю изменений данных, пока snapshot-ы не удалены при обслуживании. Schema evolution (изменение структуры) и time travel (путешествие во времени) — встроенные инструменты формата.\n", "\n", "Как устроен этот ноутбук:\n", "- Schema evolution демонстрируем на нашей рабочей таблице `silver` (это безопасная операция).\n", @@ -165,7 +167,9 @@ "\n", "Добавляем колонку `loaded_at` к silver-таблице. В PostgreSQL `ALTER TABLE ADD COLUMN` — обычная операция. В Iceberg — аналогично, но с важным свойством: Iceberg хранит историю схем.\n", "\n", - "Важно: `ALTER TABLE ADD COLUMNS` **НЕ создаёт** новый data snapshot. Это изменение метаданных (schema), а не данных: data files не перезаписываются, существующие строки логически получают NULL. Snapshot-ы фиксируют только операции с данными (INSERT, DELETE, overwrite).\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" ] }, { @@ -249,7 +253,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "Trino увидел новую колонку мгновенно. Spark изменил метаданные в PostgreSQL (JDBC catalog), Trino прочитал обновлённые метаданные. Decoupled compute + общий каталог в действии.\n" + "Trino увидел новую колонку. Spark записал новый metadata-файл в MinIO и обновил указатель на него в PostgreSQL (JDBC catalog). Trino прочитал схему из нового файла. Decoupled compute + общий каталог в действии.\n" ] }, { @@ -443,7 +447,7 @@ "source": [ "#### 7c: Time travel по времени\n", "\n", - "Альтернативный способ — по timestamp. `TIMESTAMP AS OF` — Iceberg находит ближайший snapshot, существовавший на этот момент времени. Для точного отката лучше использовать `snapshot_id`.\n" + "Альтернативный способ — по timestamp. `TIMESTAMP AS OF` читает snapshot, который был текущим в указанный момент времени. Чтобы прочитать конкретный snapshot по его ID, используй `VERSION AS OF`. Оба запроса оставляют текущее состояние таблицы без изменений.\n" ] }, { diff --git a/notebooks/08_maintenance_and_final_lab.ipynb b/notebooks/08_maintenance_and_final_lab.ipynb index b96f0c1..81c2f44 100644 --- a/notebooks/08_maintenance_and_final_lab.ipynb +++ b/notebooks/08_maintenance_and_final_lab.ipynb @@ -315,9 +315,15 @@ "source": [ "### Секция 4: expire_snapshots — очистка истории\n", "\n", - "`expire_snapshots` удаляет старые snapshot-ы из метаданных. \n", - "**Важный компромисс:** после expire_snapshots time travel к удалённым snapshot-ам невозможен. Это необратимо. \n", - "В production нужно решить: сколько истории хранить? Обычно 10–30 дней. Для демо оставим только 2 последних snapshot-а.\n" + "`expire_snapshots` удаляет старые snapshot-ы из метаданных и файлы, которые больше не нужны ни одному сохранённому snapshot-у. Очистка выполняется при вызове процедуры.\n", + "\n", + "Два параметра определяют, какие snapshot-ы будут удалены:\n", + "- `older_than` — граница времени: удалить можно snapshot-ы, созданные раньше неё. Без явной границы при стандартных настройках удаляются только snapshot-ы старше 5 дней.\n", + "- `retain_last` — сколько последних snapshot-ов в цепочке предков текущего состояния сохранить независимо от возраста. Это минимум, а не требование оставить ровно столько.\n", + "\n", + "Поэтому один `retain_last => 2` не удалит snapshot-ы, которые мы только что создали. Для демо возьмём текущее время Spark-сессии как `older_than` и сохраним два последних snapshot-а.\n", + "\n", + "**Важный компромисс:** time travel к удалённым snapshot-ам станет невозможен. Текущее время как границу используем только для одноразовой демо-таблицы; в production срок хранения выбирают с учётом требований к истории и восстановлению.\n" ] }, { @@ -328,12 +334,14 @@ "source": [ "# Snapshot-история ДО expire\n", "snapshots_before = spark.sql(f\"SELECT snapshot_id, committed_at, operation FROM {DEMO_TABLE}.snapshots\")\n", - "print(f\"Snapshot-ов до expire: {snapshots_before.count()}\")\n", + "snapshot_count_before = snapshots_before.count()\n", + "print(f\"Snapshot-ов до expire: {snapshot_count_before}\")\n", "snapshots_before.show(truncate=False)\n", "\n", "# Сохраняем ID самого старого snapshot для проверки time travel после expire\n", "first_snapshot_id = snapshots_before.orderBy(\"committed_at\").first()[\"snapshot_id\"]\n", "print(f\"Самый старый snapshot: {first_snapshot_id}\")\n", + "row_count_before_expire = spark.table(DEMO_TABLE).count()\n", "\n" ] }, @@ -343,10 +351,16 @@ "metadata": {}, "outputs": [], "source": [ - "# Выполняем expire_snapshots (оставляем только 2 последних)\n", + "# Получаем границу времени в часовом поясе Spark-сессии\n", + "cutoff = spark.sql(\"\"\"\n", + " SELECT date_format(current_timestamp(), 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS cutoff\n", + "\"\"\").first()[\"cutoff\"]\n", + "\n", + "# Удаляем snapshot-ы до cutoff, сохраняя два последних\n", "spark.sql(f\"\"\"\n", " CALL lakehouse.system.expire_snapshots(\n", " table => 'default.taxi_maintenance_demo',\n", + " older_than => TIMESTAMP '{cutoff}',\n", " retain_last => 2\n", " )\n", "\"\"\")\n", @@ -362,8 +376,16 @@ "source": [ "# Snapshot-история ПОСЛЕ expire\n", "snapshots_after = spark.sql(f\"SELECT snapshot_id, committed_at, operation FROM {DEMO_TABLE}.snapshots\")\n", - "print(f\"Snapshot-ов после expire: {snapshots_after.count()}\")\n", + "remaining_snapshot_ids = [row[\"snapshot_id\"] for row in snapshots_after.collect()]\n", + "print(f\"Snapshot-ов: {snapshot_count_before} → {len(remaining_snapshot_ids)}\")\n", "snapshots_after.show(truncate=False)\n", + "assert len(remaining_snapshot_ids) == 2, \"Ожидалось ровно 2 snapshot-а после очистки.\"\n", + "assert first_snapshot_id not in remaining_snapshot_ids, \"Самый старый snapshot не удалён.\"\n", + "\n", + "# Очистка истории не меняет данные в текущем состоянии\n", + "row_count_after_expire = spark.table(DEMO_TABLE).count()\n", + "assert row_count_after_expire == row_count_before_expire, \"Число строк изменилось после очистки.\"\n", + "print(f\"Данные на месте: {row_count_after_expire} строк.\")\n", "\n" ] }, @@ -385,10 +407,11 @@ " SELECT count(*) FROM {DEMO_TABLE}\n", " VERSION AS OF {first_snapshot_id}\n", " \"\"\").show()\n", - " print(\"Time travel сработал (snapshot не был удалён).\")\n", "except Exception as e:\n", " print(f\"Ожидаемая ошибка: {e}\")\n", " print(\"Snapshot удалён — time travel невозможен.\")\n", + "else:\n", + " raise AssertionError(\"Time travel сработал: старый snapshot не был удалён.\")\n", "\n" ] }, @@ -398,7 +421,9 @@ "source": [ "**Вывод:** expire нужно запускать обдуманно. Слишком агрессивно (потеря страховки для rollback), слишком редко (рост метаданных и стоимости хранения).\n", "\n", - "*Примечание: физическое удаление файлов из MinIO зависит от конфигурации. Иногда требуется дополнительно запускать процедуру `remove_orphan_files`, которая физически чистит файлы без привязки к активным snapshot-ам.*\n" + "В нашем демо сохранены snapshot после compaction и его предок до compaction. Предку ещё нужны старые мелкие файлы, поэтому `expire_snapshots` их не удаляет. Удаление snapshot-ов не обязательно освобождает data files: важны ссылки из оставшихся состояний.\n", + "\n", + "`remove_orphan_files` решает другую задачу: удаляет файлы, на которые вообще нет ссылок в метаданных таблицы, например остатки неудачной записи. Для файлов, нужных сохранённым snapshot-ам, она не подходит.\n" ] }, { @@ -414,11 +439,11 @@ "| 2 | `expire_snapshots` | Удаляет старые snapshot-ы | Метаданные и хранилище растут |\n", "\n", "\n", - "Если сделать наоборот: expire удалит snapshot-ы, но мелкие файлы останутся (так как на них всё ещё ссылается текущий snapshot). Compaction затем создаст новые крупные файлы, а старые мелкие станут \"orphan\" (сиротами) и не удалятся автоматически.\n", + "Если выполнить expire до compaction, мелкие файлы останутся: на них ссылается текущий snapshot. После compaction они будут нужны его предку. Удалить их сможет следующий вызов `expire_snapshots`, когда этот предок попадёт под условия очистки. До этого файлы не являются «сиротами».\n", "\n", "**Типичные ошибки:**\n", "1. **Никогда не запускать compaction:** таблица обрастает тысячами мелких файлов, чтение деградирует в десятки раз.\n", - "2. **`expire_snapshots` сразу после ошибки (с `retain_last => 1`):** если записали плохие данные и запустили expire, откатиться назад (rollback) уже не выйдет. Правило: перед expire убедись, что текущее состояние корректно.\n", + "2. **Удалить исправный snapshot после ошибочной записи:** если `older_than` и `retain_last` допускают его удаление, после expire откатиться к нему уже не выйдет. Правило: перед expire убедись, что текущее состояние корректно.\n", "3. **Обслуживание на каждый INSERT:** compaction — тяжёлая операция. В production его делают по расписанию (например, 1 раз в сутки ночью).\n" ] }, @@ -455,8 +480,8 @@ "- Создай таблицу `lakehouse.default.taxi_compaction_practice` через CTAS (500 строк из `silver`). \n", "- Выполни 5 INSERT INTO по 200 строк. \n", "- Посмотри файловую статистику ДО.\n", - "- Выполни compaction и expire_snapshots (retain_last => 1). \n", - "- Проверь результат ПОСЛЕ. \n", + "- Выполни compaction, затем получи свежий `cutoff` по образцу Секции 4 и вызови `expire_snapshots` с `older_than => TIMESTAMP '{cutoff}'` и `retain_last => 1`.\n", + "- Проверь результат ПОСЛЕ: остался 1 snapshot, число строк не изменилось.\n", "- Удали таблицу (`DROP TABLE`).\n" ] }, @@ -565,7 +590,7 @@ "\n", "**Шаг 6 (snapshot-ы).** Выполни 3 `INSERT INTO trips_silver` (например, по 300 строк из `trips_bronze`). Каждый INSERT должен включать трансформации из шага 3, плюс `current_timestamp() AS processed_at`. *Подсказка: если не указать `processed_at`, Spark выдаст ошибку несовпадения числа колонок; альтернативный вариант — использовать явный список target-колонок в `INSERT INTO trips_silver (col1, col2, ...)`.* Посмотри файловую статистику и snapshot-ы.\n", "\n", - "**Шаг 7 (обслуживание).** Выполни compaction (`rewrite_data_files`), затем `expire_snapshots` (retain_last => 1). Проверь, что файлов стало меньше, а snapshot-ов остался 1.\n", + "**Шаг 7 (обслуживание).** Выполни compaction (`rewrite_data_files`). После него получи свежий `cutoff` по образцу Секции 4 и вызови `expire_snapshots` с `older_than => TIMESTAMP '{cutoff}'` и `retain_last => 1`. Проверь, что файлов стало меньше, snapshot-ов остался 1, а число строк не изменилось.\n", "\n", "**Шаг 8 (cleanup).** Удали `trips_silver`, `trips_bronze` и namespace `lakehouse.final`.\n", "\n"