ddadmin
bf3241124c
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.
38 KiB
38 KiB
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-08-maintenance-and-lab") \
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
print("SparkSession создана.")
SILVER_TABLE = "lakehouse.silver.nyc_taxi_yellow"
DEMO_TABLE = "lakehouse.default.taxi_maintenance_demo"
RAW_PATH = "s3a://lakehouse/raw/nyc_taxi/yellow_tripdata_*.parquet"
In [ ]:
# Проверяем, что silver-таблица существует (Модуль 5 должен быть пройден)
assert spark.catalog.tableExists(SILVER_TABLE), f"Таблица {SILVER_TABLE} не найдена. Пройдите Модуль 5."
assert spark.table(SILVER_TABLE).count() > 0, f"Таблица {SILVER_TABLE} пуста."
print(f"Таблица {SILVER_TABLE} готова к работе.")
# Проверяем, что raw-данные доступны (Модуль 3)
try:
spark.read.parquet(RAW_PATH).limit(1).collect()
print(f"Raw-данные доступны по пути: {RAW_PATH}")
except Exception as e:
raise AssertionError(f"Raw-данные не найдены по пути {RAW_PATH}. Вернись в Модуль 3 и выполни загрузку данных в MinIO.") from e
In [ ]:
# Helper для выполнения запросов к Trino
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 функция готова.")
In [ ]:
# Демо-таблица одноразовая: удаляем данные и старую snapshot-историю,
# чтобы повторный запуск дал те же 9 snapshot-ов.
spark.sql(f"DROP TABLE IF EXISTS {DEMO_TABLE} PURGE")
# Создание демо-таблицы (создаст 1 файл и 1 snapshot)
spark.sql(f"""
CREATE 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 500
""")
print(f"Демо-таблица создана: {spark.table(DEMO_TABLE).count()} строк.")
# 8 мелких INSERT INTO (создадут 8 новых мелких файлов)
print("Запуск 8 INSERT-ов (может занять 1-2 минуты)...")
for i in range(8):
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}
LIMIT 100
""")
print(f"INSERT {i+1}/8 выполнен.")
print(f"Итого строк: {spark.table(DEMO_TABLE).count()}")
In [ ]:
# Файловая статистика через metadata-таблицу files
files_df = spark.sql(f"""
SELECT file_path, file_format, record_count, file_size_in_bytes
FROM {DEMO_TABLE}.files
""")
print(f"Количество data files: {files_df.count()}")
files_df.show(truncate=40)
In [ ]:
# Snapshot-история
spark.sql(f"SELECT snapshot_id, committed_at, operation, summary FROM {DEMO_TABLE}.snapshots").show(truncate=False)
In [ ]:
silver_files = spark.sql(f"SELECT file_path, record_count, file_size_in_bytes FROM {SILVER_TABLE}.files")
print(f"Silver: {silver_files.count()} data file(s)")
silver_files.show(truncate=40)
In [ ]:
# Сохраним статистику ДО compaction
files_before = spark.sql(f"SELECT count(*) AS file_count, sum(file_size_in_bytes) AS total_bytes FROM {DEMO_TABLE}.files").first()
print(f"До compaction: {files_before['file_count']} файлов, {files_before['total_bytes']} байт")
In [ ]:
# Выполняем compaction
spark.sql(f"CALL lakehouse.system.rewrite_data_files(table => 'default.taxi_maintenance_demo')")
print("Compaction выполнен.")
In [ ]:
# Статистика ПОСЛЕ compaction
files_after = spark.sql(f"SELECT count(*) AS file_count, sum(file_size_in_bytes) AS total_bytes FROM {DEMO_TABLE}.files").first()
print(f"После compaction: {files_after['file_count']} файлов, {files_after['total_bytes']} байт")
print(f"Было {files_before['file_count']} файлов → стало {files_after['file_count']} файлов")
# Проверим, что данные на месте
print(f"Строк в таблице: {spark.table(DEMO_TABLE).count()}")
In [ ]:
# Посмотрим snapshot-историю: появился новый snapshot
spark.sql(f"SELECT snapshot_id, committed_at, operation FROM {DEMO_TABLE}.snapshots").show(truncate=False)
In [ ]:
# Snapshot-история ДО expire
snapshots_before = spark.sql(f"SELECT snapshot_id, committed_at, operation FROM {DEMO_TABLE}.snapshots")
snapshot_count_before = snapshots_before.count()
print(f"Snapshot-ов до expire: {snapshot_count_before}")
snapshots_before.show(truncate=False)
# Сохраняем ID самого старого snapshot для проверки time travel после expire
first_snapshot_id = snapshots_before.orderBy("committed_at").first()["snapshot_id"]
print(f"Самый старый snapshot: {first_snapshot_id}")
row_count_before_expire = spark.table(DEMO_TABLE).count()
In [ ]:
# Получаем границу времени в часовом поясе Spark-сессии
cutoff = spark.sql("""
SELECT date_format(current_timestamp(), 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS cutoff
""").first()["cutoff"]
# Удаляем snapshot-ы до cutoff, сохраняя два последних
spark.sql(f"""
CALL lakehouse.system.expire_snapshots(
table => 'default.taxi_maintenance_demo',
older_than => TIMESTAMP '{cutoff}',
retain_last => 2
)
""")
print("expire_snapshots выполнен.")
In [ ]:
# Snapshot-история ПОСЛЕ expire
snapshots_after = spark.sql(f"SELECT snapshot_id, committed_at, operation FROM {DEMO_TABLE}.snapshots")
remaining_snapshot_ids = [row["snapshot_id"] for row in snapshots_after.collect()]
print(f"Snapshot-ов: {snapshot_count_before} → {len(remaining_snapshot_ids)}")
snapshots_after.show(truncate=False)
assert len(remaining_snapshot_ids) == 2, "Ожидалось ровно 2 snapshot-а после очистки."
assert first_snapshot_id not in remaining_snapshot_ids, "Самый старый snapshot не удалён."
# Очистка истории не меняет данные в текущем состоянии
row_count_after_expire = spark.table(DEMO_TABLE).count()
assert row_count_after_expire == row_count_before_expire, "Число строк изменилось после очистки."
print(f"Данные на месте: {row_count_after_expire} строк.")
In [ ]:
try:
spark.sql(f"""
SELECT count(*) FROM {DEMO_TABLE}
VERSION AS OF {first_snapshot_id}
""").show()
except Exception as e:
print(f"Ожидаемая ошибка: {e}")
print("Snapshot удалён — time travel невозможен.")
else:
raise AssertionError("Time travel сработал: старый snapshot не был удалён.")
In [ ]:
# Ваш код: файловая статистика silver (файлы, размеры, snapshot-ы)
In [ ]:
# Ваш код: создание taxi_compaction_practice + 5 INSERT
In [ ]:
# Ваш код: файловая статистика до compaction
In [ ]:
# Ваш код: compaction + expire_snapshots
In [ ]:
# Ваш код: файловая статистика после + DROP TABLE
In [ ]:
spark.sql(f"DROP TABLE IF EXISTS {DEMO_TABLE} PURGE")
print("Демо-таблица и её учебные данные удалены.")
In [ ]:
# Шаг 1: CREATE NAMESPACE
In [ ]:
# Шаг 2: прочитай raw и создай trips_bronze (LIMIT 5000)
In [ ]:
# Шаг 2: проверка trips_bronze (count, printSchema)
In [ ]:
# Шаг 3: создай trips_silver с трансформациями
In [ ]:
# Шаг 3: проверка trips_silver (count, проверка фильтра)
In [ ]:
# Шаг 4: чтение trips_silver из Trino
In [ ]:
# Шаг 5: ALTER TABLE ADD COLUMNS
In [ ]:
# Шаг 5: проверка из Trino (DESCRIBE)
In [ ]:
# Шаг 6: 3 INSERT INTO trips_silver
In [ ]:
# Шаг 6: файловая статистика и snapshot-ы
In [ ]:
# Шаг 7: compaction + expire_snapshots
In [ ]:
# Шаг 7: проверка результата
In [ ]:
# Шаг 8: cleanup (DROP TABLE, DROP NAMESPACE)
In [ ]:
spark.stop()