mini-lakehouse-lab/notebooks/07_safe_table_changes.ipynb

Модуль 7. Безопасная работа с таблицами: schema evolution и time travel

Цель: Научиться безопасно изменять Iceberg-таблицы, использовать встроенные механизмы time travel (чтение предыдущих состояний) и rollback (восстановление после ошибок).

В этом модуле мы сравним подходы классических DWH и Lakehouse к работе с изменениями:

	Классический DWH (Greenplum)	Lakehouse (Iceberg)
Добавить колонку	ALTER TABLE ADD COLUMN (мгновенно, NULL)	ALTER TABLE ADD COLUMNS (мгновенно, NULL)
Переименовать колонку	ALTER TABLE RENAME COLUMN	ALTER TABLE RENAME COLUMN
Откатить данные к вчерашнему состоянию	Восстановление из бэкапа (pg_dump / PITR)	SELECT ... VERSION AS OF <snapshot_id>
Посмотреть историю изменений таблицы	Нет встроенного механизма	SELECT * FROM table.snapshots
Последствия ошибочной записи	Нужен бэкап или ручной откат	Rollback к предыдущему snapshot

Ключевая идея: Iceberg хранит полную историю изменений данных. Schema evolution (изменение структуры) и time travel (путешествие во времени) — встроенные инструменты формата, а не дополнительная инфраструктура.

Как устроен этот ноутбук:

• Schema evolution демонстрируем на нашей рабочей таблице silver (это безопасная операция).
• Time travel и восстановление после ошибки — на отдельной демо-таблице (чтобы не рисковать данными для Модуля 8).
• Trino используется для проверки эффекта на downstream (системы, читающие данные после нас).

Секция 1: Spark-сессия, Trino-клиент и проверка таблиц

In [ ]:

import pandas as pd
from pyspark.sql import SparkSession
from trino.dbapi import connect
import trino
from IPython.display import display

spark = SparkSession.builder \
    .appName("module-07-safe-table-changes") \
    .getOrCreate()

spark.sparkContext.setLogLevel("ERROR")
print("SparkSession создана.")

SILVER_TABLE = "lakehouse.silver.nyc_taxi_yellow"
DEMO_TABLE = "lakehouse.default.taxi_changes_demo"

In [ ]:

# Проверяем, что silver-таблица существует (Модуль 5 должен быть пройден)
assert spark.catalog.tableExists(SILVER_TABLE), f"Таблица {SILVER_TABLE} не найдена. Пройдите Модуль 5."
count = spark.table(SILVER_TABLE).count()
assert count > 0, f"Таблица {SILVER_TABLE} пуста."
print(f"Таблица {SILVER_TABLE} готова к работе. Строк: {count}")

In [ ]:

def trino_query(sql: str) -> pd.DataFrame:
    """Выполняет SQL-запрос в Trino и возвращает результат как Pandas DataFrame."""
    conn = connect(
        host="trino",
        port=8080,
        user="jupyter",
        catalog="lakehouse"
    )
    cur = conn.cursor()
    try:
        cur.execute(sql)
        rows = cur.fetchall()
        columns = [desc[0] for desc in cur.description]
        return pd.DataFrame(rows, columns=columns)
    except trino.exceptions.ProgrammingError as e:
        if "No nodes available to run query" in str(e):
            print("Ошибка: Trino ещё не готов. Подождите пару минут после старта контейнеров.")
            raise
        # Запросы типа CREATE/DROP не возвращают строк
        return pd.DataFrame([{"status": "Success"}])
    finally:
        conn.close()

print("Trino helper функция готова.")

Если ячейки выше упали с ImportError: trino — нужно пересобрать образ Jupyter (в терминале: docker compose build jupyter && docker compose up -d).

Silver на месте, Trino готов. В этом модуле мы будем изменять таблицу и наблюдать последствия из обоих движков.

Секция 2: Snapshot-ы — встроенная история изменений

В Модуле 4 мы впервые увидели snapshot-ы bronze-таблицы. Теперь разберёмся глубже. Каждая операция с данными в Iceberg (INSERT, overwrite, delete) автоматически создаёт snapshot — «снимок» состояния таблицы. Snapshot фиксирует, какие data files составляли таблицу в этот момент. Это как коммит в git: можно вернуться к любому предыдущему состоянию.

In [ ]:

# Посмотрим snapshot-историю silver-таблицы
spark.sql(f"SELECT snapshot_id, committed_at, operation, summary FROM {SILVER_TABLE}.snapshots").show(truncate=False)

In [ ]:

# Дополнительная история с указанием предков (родительских snapshot-ов)
spark.sql(f"SELECT * FROM {SILVER_TABLE}.history").show(truncate=False)

Поля таблицы snapshots:

• committed_at — когда произошла операция.
• operation — тип операции (append, overwrite).
• summary — краткая статистика (added-data-files, total-records и т.д.).

Если вы не перезапускали Модуль 5 через CREATE OR REPLACE, у silver будет 1 snapshot (от CTAS). В Greenplum такой истории нет: после INSERT предыдущее состояние таблицы недоступно без бэкапа.

Секция 3: Schema evolution — безопасное добавление колонки

Добавляем колонку loaded_at к silver-таблице. В PostgreSQL ALTER TABLE ADD COLUMN — обычная операция. В Iceberg — аналогично, но с важным свойством: Iceberg хранит историю схем.

Важно: ALTER TABLE ADD COLUMNS НЕ создаёт новый data snapshot. Это изменение метаданных (schema), а не данных: data files не перезаписываются, существующие строки логически получают NULL. Snapshot-ы фиксируют только операции с данными (INSERT, DELETE, overwrite).

In [ ]:

try:
    spark.sql(f"ALTER TABLE {SILVER_TABLE} ADD COLUMNS (loaded_at TIMESTAMP)")
    print("Колонка loaded_at добавлена.")
except Exception as e:
    if "already exists" in str(e).lower():
        print("Колонка loaded_at уже существует (повторный запуск ноутбука). Продолжаем.")
    else:
        raise

In [ ]:

# Проверим схему из Spark
spark.table(SILVER_TABLE).printSchema()

In [ ]:

# Посмотрим данные (должен быть NULL)
spark.sql(f"SELECT VendorID, loaded_at FROM {SILVER_TABLE} LIMIT 5").show()

Колонка добавлена. Данные не изменились — Iceberg не перезаписывал data files. В рабочем сценарии loaded_at заполнялся бы при следующих INSERT-ах: INSERT INTO ... SELECT ..., current_timestamp() AS loaded_at FROM .... Существующие строки остаются с NULL — это нормальная практика эволюции схемы. Заполнение существующих строк (UPDATE) — отдельная тема, требующая перезаписи файлов.

Примечание: если вы повторно запустите Модуль 5 после Модуля 7, CREATE OR REPLACE TABLE пересоздаст silver без колонки loaded_at. Это ещё одна иллюстрация того, почему CREATE OR REPLACE опасен в рабочих сценариях — он стирает все изменения, включая schema evolution.

Секция 4: Downstream-эффект — Trino видит изменение

В Модуле 6 мы убедились: Spark и Trino читают одну таблицу через общий каталог. Schema evolution — ещё одно следствие этого дизайна: изменение схемы в Spark мгновенно видно в Trino. Не нужно «синхронизировать» или «обновлять» что-то на стороне Trino.

В DBeaver: DESCRIBE lakehouse.silver.nyc_taxi_yellow;

In [ ]:

# Проверим схему из Trino
display(trino_query(f"DESCRIBE {SILVER_TABLE}"))

In [ ]:

# Прочитаем данные из Trino
display(trino_query(f"SELECT vendorid, loaded_at FROM {SILVER_TABLE} LIMIT 5"))

Trino увидел новую колонку мгновенно. Spark изменил метаданные в PostgreSQL (JDBC catalog), Trino прочитал обновлённые метаданные. Decoupled compute + общий каталог в действии.

Секция 5: Обзор операций schema evolution

ADD COLUMN — самая безопасная операция. Но Iceberg поддерживает и другие:

Операция	Spark SQL	Безопасность	Влияние на downstream
Добавить колонку	ALTER TABLE ADD COLUMNS (col type)	Безопасно	Новая колонка с NULL, существующие запросы не ломаются
Переименовать колонку	ALTER TABLE RENAME COLUMN old TO new	Осторожно	Запросы с SELECT old_name перестают работать
Расширить тип	ALTER TABLE ALTER COLUMN col TYPE bigint	Безопасно	INT -> BIGINT: без потери данных
Сузить тип	—	Опасно	BIGINT -> INT: возможна потеря данных. Iceberg не поддерживает
Удалить колонку	ALTER TABLE DROP COLUMN col	Опасно	Запросы с SELECT col перестают работать

Далее в модуле мы продемонстрируем RENAME COLUMN на демо-таблице и покажем, как это ломает downstream-запросы.

Секция 6: Подготовка демо-таблицы с несколькими snapshot-ами

Для экспериментов с time travel и восстановлением (rollback) создаём отдельную демо-таблицу. Silver не трогаем — он нужен для финального Модуля 8.

Мы создадим таблицу с небольшим подмножеством silver (1000 строк) и нарастим snapshot-историю через дополнительные INSERT INTO.

In [ ]:

# Создание демо-таблицы
spark.sql(f"""
    CREATE OR REPLACE TABLE {DEMO_TABLE}
    USING iceberg AS
    SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,
           passenger_count, trip_distance, fare_amount, total_amount,
           pickup_borough, pickup_zone
    FROM {SILVER_TABLE}
    LIMIT 1000
""")

count_1 = spark.table(DEMO_TABLE).count()
assert count_1 == 1000, f"Ожидалось 1000 строк, получено {count_1}"
print(f"Таблица создана. Строк: {count_1}")

In [ ]:

# Сохраним ID первого snapshot-а
df_snapshots = spark.sql(f"SELECT snapshot_id, committed_at FROM {DEMO_TABLE}.snapshots ORDER BY committed_at ASC").collect()
assert len(df_snapshots) == 1, f"Ожидался 1 snapshot, получено {len(df_snapshots)}"
snapshot_1 = df_snapshots[0]['snapshot_id']
committed_at_1 = df_snapshots[0]['committed_at']

print(f"Snapshot 1 ID: {snapshot_1} (создан: {committed_at_1})")

In [ ]:

# Добавим ещё 500 строк из Манхэттена (создаст второй snapshot)
spark.sql(f"""
    INSERT INTO {DEMO_TABLE}
    SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,
           passenger_count, trip_distance, fare_amount, total_amount,
           pickup_borough, pickup_zone
    FROM {SILVER_TABLE}
    WHERE pickup_borough = 'Manhattan'
    LIMIT 500
""")

count_2 = spark.table(DEMO_TABLE).count()
print(f"Добавлено 500 строк. Всего строк: {count_2}")

In [ ]:

# Сохраним ID второго snapshot-а
df_snapshots = spark.sql(f"SELECT snapshot_id, committed_at FROM {DEMO_TABLE}.snapshots ORDER BY committed_at ASC").collect()
snapshot_2 = df_snapshots[1]['snapshot_id']
committed_at_2 = df_snapshots[1]['committed_at']

print(f"Snapshot 2 ID: {snapshot_2} (создан: {committed_at_2})")

Теперь у таблицы два snapshot-а: начальная загрузка (1000 строк) и добавление (500 строк). Это как два коммита в git. Каждый фиксирует конкретное состояние таблицы.

Секция 7: Time travel — чтение предыдущих состояний

Центральная возможность Iceberg: можно прочитать таблицу в том состоянии, в каком она была на момент любого snapshot-а. Это time travel. В Greenplum для этого пришлось бы восстанавливать бэкап целой базы или настраивать сложный PITR.

7a: Time travel через Spark

In [ ]:

# Чтение первого snapshot-а (начальная загрузка)
print(f"Читаем таблицу по состоянию Snapshot 1 ({snapshot_1}):")

spark.sql(f"""
    SELECT count(*) AS row_count
    FROM {DEMO_TABLE}
    VERSION AS OF {snapshot_1}
""").show()

In [ ]:

# Чтение текущего состояния для сравнения
print("Читаем текущее состояние таблицы:")

spark.sql(f"""
    SELECT count(*) AS row_count
    FROM {DEMO_TABLE}
""").show()

Данные из первого snapshot-а не потеряны — оба состояния доступны одновременно. Iceberg хранит все data files; snapshot определяет, какие из них составляют таблицу в конкретный момент.

7b: Time travel через Trino

Trino тоже умеет time travel. Тот же snapshot, та же таблица, тот же каталог. Обратите внимание на синтаксис: в Spark — VERSION AS OF, в Trino — FOR VERSION AS OF.

В DBeaver: SELECT count(*) FROM lakehouse.default.taxi_changes_demo FOR VERSION AS OF <snapshot_id>;

In [ ]:

# Time travel через Trino к первому snapshot-у
print(f"Trino читает Snapshot 1 ({snapshot_1}):")

display(trino_query(f"""
    SELECT count(*) AS row_count
    FROM {DEMO_TABLE}
    FOR VERSION AS OF {snapshot_1}
"""))

7c: Time travel по времени

Альтернативный способ — по timestamp. TIMESTAMP AS OF — Iceberg находит ближайший snapshot, существовавший на этот момент времени. Для точного отката лучше использовать snapshot_id.

In [ ]:

# Time travel по времени (состояние на момент создания Snapshot 1)
spark.sql(f"""
    SELECT count(*) AS row_count
    FROM {DEMO_TABLE}
    TIMESTAMP AS OF '{committed_at_1}'
""").show()

Секция 8: Намеренная ошибка и восстановление (Rollback)

Ключевой урок модуля. Мы намеренно внесём «плохие» данные, затем восстановим таблицу через rollback. Ошибки в Iceberg — не катастрофа, если понимаешь snapshot-историю.

8a: Вносим ошибку

In [ ]:

# INSERT «плохих» данных (200 строк с отрицательным fare_amount)
spark.sql(f"""
    INSERT INTO {DEMO_TABLE}
    SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,
           passenger_count, trip_distance,
           -99.99 AS fare_amount,
           -99.99 AS total_amount,
           'ERROR' AS pickup_borough,
           'BAD_DATA' AS pickup_zone
    FROM {SILVER_TABLE}
    LIMIT 200
""")

# Убедимся, что таблица "испорчена"
spark.sql(f"""
    SELECT count(*) as bad_rows 
    FROM {DEMO_TABLE} 
    WHERE fare_amount < 0
""").show()

count_3 = spark.table(DEMO_TABLE).count()
print(f"Всего строк сейчас: {count_3}")

In [ ]:

# Посмотрим на 3-й snapshot (нашу ошибку)
spark.sql(f"SELECT snapshot_id, operation FROM {DEMO_TABLE}.snapshots").show(truncate=False)

В Greenplum: 200 ошибочных строк в production-таблице -> звонок DBA, восстановление из бэкапа (если он есть и свежий), простой сервиса, или опасные ручные DELETE/UPDATE без возможности отката. В Iceberg: rollback к предыдущему snapshot-у.

8b: Восстановление через rollback

Мы откатимся к snapshot 2, который мы сохранили в переменной snapshot_2.

In [ ]:

# Выполняем rollback
spark.sql(f"""
    CALL lakehouse.system.rollback_to_snapshot(
        table => 'default.taxi_changes_demo',
        snapshot_id => {snapshot_2}
    )
""")
print(f"Rollback к snapshot {snapshot_2} завершен.")

In [ ]:

# Проверка восстановления
bad_rows = spark.sql(f"SELECT count(*) as bad_rows FROM {DEMO_TABLE} WHERE fare_amount < 0").collect()[0]['bad_rows']
current_count = spark.table(DEMO_TABLE).count()

print(f"Плохих строк (fare_amount < 0): {bad_rows}")
print(f"Всего строк: {current_count}")

Таблица восстановлена!

8c: Что произошло со snapshot-историей?

In [ ]:

spark.sql(f"SELECT snapshot_id, operation FROM {DEMO_TABLE}.snapshots").show(truncate=False)

Теперь в истории 4 записи. Четвёртый snapshot — это результат rollback, но его данные (набор файлов) в точности соответствуют snapshot 2.

Rollback не удаляет историю и не удаляет файлы. Он создаёт новый snapshot, указывающий на данные предыдущего состояния. Аналогия:

	git	Iceberg
Откат с сохранением истории	git revert (новый коммит)	rollback_to_snapshot (новый snapshot)
Откат с удалением истории	git reset --hard	CREATE OR REPLACE (уничтожает snapshot-ы)

Старые data files (включая файлы с «плохими» строками) остаются в хранилище (MinIO) до процедуры expire_snapshots — это тема Модуля 8. Rollback не чистит storage, он только меняет указатель текущего snapshot-а.

Секция 9: RENAME COLUMN и влияние на downstream

Покажем, что RENAME COLUMN — безопасная операция для данных (они не перезаписываются), но опасная для downstream-запросов (они могут сломаться).

In [ ]:

# Переименуем колонку в демо-таблице
spark.sql(f"ALTER TABLE {DEMO_TABLE} RENAME COLUMN pickup_borough TO borough")
print("Колонка переименована в Spark.")

In [ ]:

# Посмотрим схему из Spark - колонка теперь borough
spark.table(DEMO_TABLE).printSchema()

In [ ]:

# Запрос с новым именем borough через Trino работает
display(trino_query(f"SELECT borough FROM {DEMO_TABLE} LIMIT 3"))

In [ ]:

# А запрос со старым именем pickup_borough упадет с ошибкой
try:
    display(trino_query(f"SELECT pickup_borough FROM {DEMO_TABLE} LIMIT 3"))
except Exception as e:
    print(f"Ошибка в Trino: {e}")

Данные не потеряны — колонка переименована в метаданных. Но любой downstream-запрос, использовавший старое имя pickup_borough, сломается. В рабочем окружении перед RENAME нужно проверить, кто и как читает эту колонку.

Секция 10: Типичные ошибки новичка

Ошибка 1: Слепой CREATE OR REPLACE. В Модулях 4 и 5 мы использовали CREATE OR REPLACE TABLE ... AS SELECT (CTAS) для создания bronze и silver. Это был осознанный компромисс: первичная загрузка, нет предыдущей истории, которую нужно сохранять. Но в рабочих сценариях CREATE OR REPLACE — деструктивная операция: она уничтожает ВСЮ snapshot-историю таблицы. Если у таблицы было 100 snapshot-ов, после CREATE OR REPLACE останется один. Time travel к предыдущим состояниям станет невозможен.

Ошибка 2: Изменение схемы без проверки downstream. Мы только что видели, как RENAME COLUMN ломает запросы со старым именем. ALTER COLUMN TYPE (расширение INT -> BIGINT) безопасен для данных, но downstream-системы могут интерпретировать новый тип по-другому и сломаться на своей стороне. Перед любым изменением схемы нужно понимать, кто читает эту таблицу.

Ошибка 3: Отсутствие проверки snapshot-истории перед деструктивной операцией. Перед любой сложной операцией, массово изменяющей данные (UPDATE/DELETE), полезно посмотреть текущее состояние: SELECT * FROM table.snapshots. Это занимает секунды и даёт понимание, какой snapshot станет точкой отката в случае проблемы. Привычка: посмотрел snapshot-ы -> выполнил операцию -> проверил результат -> убедился, что новый snapshot создан корректно.

Секция 11: Самостоятельное задание

Пересоздадим демо-таблицу с чистого листа. После демонстраций выше её схема изменена, а история содержит учебные эксперименты. Для самостоятельной работы нужна чистая таблица.

In [ ]:

# Пересоздаем демо-таблицу (это удалит всю ее snapshot-историю - Ошибка 1 в действии!)
spark.sql(f"""
    CREATE OR REPLACE TABLE {DEMO_TABLE}
    USING iceberg AS
    SELECT VendorID, tpep_pickup_datetime, tpep_dropoff_datetime,
           passenger_count, trip_distance, fare_amount, total_amount,
           pickup_borough, pickup_zone
    FROM {SILVER_TABLE}
    LIMIT 1000
""")
print(f"Демо-таблица пересоздана начисто: {spark.table(DEMO_TABLE).count()} строк, 1 snapshot.")

Выполни 5 задач на демо-таблице lakehouse.default.taxi_changes_demo.

Задача 1 (schema evolution). Добавь колонку quality_flag STRING к демо-таблице через ALTER TABLE ADD COLUMNS. Проверь из Spark и из Trino, что колонка появилась и все значения NULL.

DBeaver: SELECT * FROM lakehouse.default.taxi_changes_demo LIMIT 10;

In [ ]:

# Ваш код: ALTER TABLE ADD COLUMNS (quality_flag)

In [ ]:

# Ваш код: проверка из Spark и Trino

Задача 2 (создание snapshot). Вставь (INSERT INTO) 300 строк из silver (с условием pickup_borough = 'Brooklyn') в демо-таблицу. Проверь, что появился новый snapshot (всего их станет 2).

In [ ]:

# Ваш код: INSERT 300 строк из Brooklyn

Задача 3 (time travel). Посмотри snapshot-историю демо-таблицы. Прочитай таблицу в состоянии ДО добавления 300 строк (т.е. к snapshot 1) через VERSION AS OF. Сравни количество строк.

In [ ]:

# Ваш код: snapshot-история

In [ ]:

# Ваш код: time travel — VERSION AS OF

Задача 4 (recovery). Сделай INSERT 100 строк с fare_amount = -1 и pickup_zone = 'STUDENT_ERROR' в демо-таблицу. Убедись, что «плохие» строки появились. Затем выполни rollback_to_snapshot к snapshot-у ДО этого INSERT-а (т.е. к snapshot 2). Проверь, что таблица восстановлена.

In [ ]:

# Ваш код: INSERT 100 «плохих» строк

In [ ]:

# Ваш код: rollback_to_snapshot

In [ ]:

# Ваш код: проверка восстановления

Задача 5 (ответ). Ответь текстом в ячейке ниже: чем rollback_to_snapshot отличается от CREATE OR REPLACE TABLE ... AS SELECT * FROM table VERSION AS OF <snapshot_id>? Что происходит с историей snapshot-ов в каждом случае?

Ваш ответ на Задачу 5: ...

(Дополнительно, по желанию): Посмотри snapshot-историю silver-таблицы. Сколько snapshot-ов у неё? Какие операции их создали?

In [ ]:

# Ваш код: snapshot-история silver

Секция 12: Checkpoint

Ответь для себя на вопросы:

1. Что такое snapshot в Iceberg и когда он создаётся?
2. Создаёт ли ALTER TABLE ADD COLUMNS новый data snapshot? Почему?
3. Чем ADD COLUMN отличается от RENAME COLUMN по влиянию на downstream?
4. Как прочитать предыдущее состояние таблицы? (назови два способа)
5. Что делает rollback_to_snapshot? Теряется ли при этом snapshot-история?
6. Почему CREATE OR REPLACE опаснее, чем INSERT INTO с последующим rollback?
7. Какой аналог time travel в классических СУБД (PostgreSQL/Greenplum) и почему он сложнее в использовании?

Секция 13: Завершение

Удаляем демо-таблицу — она больше не нужна. Silver остаётся для Модуля 8 (колонка loaded_at — демонстрационная; Модуль 8 не зависит от неё).

In [ ]:

spark.sql(f"DROP TABLE IF EXISTS {DEMO_TABLE}")
print("Демо-таблица удалена.")

Итог модуля:

• Schema evolution (ADD COLUMN) — безопасная metadata-only операция, которая мгновенно видна из всех движков через общий каталог.
• Snapshot-ы — автоматическая история изменений данных, встроенная в Iceberg.
• Time travel — чтение предыдущих состояний без бэкапов.
• Rollback — восстановление после ошибки с сохранением истории.

В финальном Модуле 8 мы изучим базовое обслуживание таблиц: compaction (решение проблемы множества мелких файлов) и expire_snapshots (очистка старых файлов данных, оставшихся после time travel и rollback, которые мы научились применять в этом модуле).

In [ ]:

spark.stop()