ddadmin
4bea2af9d2
- Зачем: - обучение студентов базовому обслуживанию Iceberg-таблиц (compaction, expire_snapshots) и закрепление навыков всего курса. - Что: - создан notebooks/08_maintenance_and_final_lab.ipynb с обзором проблемы мелких файлов и её решения. - добавлена финальная сквозная практика (raw -> bronze -> silver -> Trino -> schema evolution -> обслуживание) в namespace lakehouse.final. - в финальную практику включены педагогические подсказки и конкретные задания (trip_duration_minutes). - статус модуля 8 в планах обновлен до Ready for validation. - Проверка: - ручная проверка выполнения ячеек в ноутбуке и синтаксиса stored procedures Iceberg.
34 KiB
34 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 [ ]:
# Создание демо-таблицы (создаст 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 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")
print(f"Snapshot-ов до expire: {snapshots_before.count()}")
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}")
In [ ]:
# Выполняем expire_snapshots (оставляем только 2 последних)
spark.sql(f"""
CALL lakehouse.system.expire_snapshots(
table => 'default.taxi_maintenance_demo',
retain_last => 2
)
""")
print("expire_snapshots выполнен.")
In [ ]:
# Snapshot-история ПОСЛЕ expire
snapshots_after = spark.sql(f"SELECT snapshot_id, committed_at, operation FROM {DEMO_TABLE}.snapshots")
print(f"Snapshot-ов после expire: {snapshots_after.count()}")
snapshots_after.show(truncate=False)
In [ ]:
try:
spark.sql(f"""
SELECT count(*) FROM {DEMO_TABLE}
VERSION AS OF {first_snapshot_id}
""").show()
print("Time travel сработал (snapshot не был удалён).")
except Exception as e:
print(f"Ожидаемая ошибка: {e}")
print("Snapshot удалён — time travel невозможен.")
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}")
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()