ddadmin
91a5c85469
- Зачем: модули 2–8 в новом процессе выбирали local[*] вместо Spark-кластера. - Что: master перенесён в общий spark-defaults.conf, smoke требует точный адрес кластера, документация и планы синхронизированы. - Проверка: фактическая инициализация всех восьми ноутбуков, повторный getOrCreate, распределённые вычисления, Iceberg/MinIO write-read-cleanup и docker compose config прошли.
13 KiB
13 KiB
In [ ]:
# Адрес Spark-кластера и настройки Iceberg берутся из spark-defaults.conf.
# Поэтому в ноутбуке достаточно задать имя приложения.
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("Module-02-Mental-Model") \
.getOrCreate()
# Spark выводит много INFO/WARN логов — убираем их, чтобы не засорять вывод ноутбука
spark.sparkContext.setLogLevel("ERROR")
print("Spark сессия готова!")In [ ]:
# Создаем namespace (логическую базу данных)
spark.sql("CREATE NAMESPACE IF NOT EXISTS lakehouse.module_02")
# Создаем или перезаписываем таблицу и вставляем пару строк
spark.sql("""
CREATE OR REPLACE TABLE lakehouse.module_02.demo_table (
id int,
name string,
created_at timestamp
)
USING iceberg
""")
spark.sql("""
INSERT INTO lakehouse.module_02.demo_table
VALUES
(1, 'Alice', current_timestamp()),
(2, 'Bob', current_timestamp())
""")
# Прочитаем данные через Spark, чтобы убедиться, что они записались
spark.sql("SELECT * FROM lakehouse.module_02.demo_table").show()In [ ]:
import psycopg2
import pandas as pd
import warnings
# Подключаемся к PostgreSQL, который хранит метаданные Iceberg (наш Catalog)
with psycopg2.connect(
host="postgres-iceberg",
database="iceberg",
user="iceberg",
password="iceberg"
) as conn:
# Запрашиваем информацию о нашей таблице
query = """
SELECT table_namespace, table_name, metadata_location, previous_metadata_location
FROM iceberg_tables
WHERE table_namespace = 'module_02'
"""
# Используем catch_warnings локально, чтобы скрыть UserWarning от pandas при чтении из DB-API
with warnings.catch_warnings():
warnings.simplefilter('ignore', UserWarning)
df_catalog = pd.read_sql(query, conn)
# Посмотрим, куда указывает catalog (обратите внимание на metadata_location)
df_catalogIn [ ]:
import boto3
# Подключаемся к MinIO
s3 = boto3.client(
's3',
endpoint_url='http://minio:9000',
aws_access_key_id='minioadmin',
aws_secret_access_key='minioadmin'
)
bucket_name = 'lakehouse'
# Путь (prefix) начинается с warehouse, так как это задано в настройках каталога
prefix = 'warehouse/module_02/demo_table/'
# Получаем список всех файлов (объектов), относящихся к нашей таблице
response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
print("Файлы таблицы в MinIO:\n")
contents = response.get('Contents', [])
if not contents:
print("Файлы не найдены. Проверьте prefix или убедитесь, что таблица создана.")
else:
for obj in contents:
key = obj['Key']
size_kb = obj['Size'] / 1024
if '/data/' in key:
print(f"[DATA] {key} ({size_kb:.2f} KB)")
elif '/metadata/' in key:
print(f"[META] {key} ({size_kb:.2f} KB)")
else:
print(f"[OTHER] {key} ({size_kb:.2f} KB)")In [ ]:
# Ваш код: создание таблицы my_first_table и вставка данных
In [ ]:
# Ваш код: чтение данных через Spark
In [ ]:
# Ваш код: инспекция каталога (PostgreSQL)
In [ ]:
# Ваш код: инспекция хранилища (MinIO)
In [ ]:
# PURGE удаляет одноразовые таблицы и из каталога, и из хранилища
spark.sql("DROP TABLE IF EXISTS lakehouse.module_02.demo_table PURGE")
spark.sql("DROP TABLE IF EXISTS lakehouse.module_02.my_first_table PURGE")
# Удаляем namespace
spark.sql("DROP NAMESPACE IF EXISTS lakehouse.module_02")
spark.stop()
print("Очистка завершена, Spark сессия остановлена.")