Hive Metastore: external vs managed tables и partitioned tables DDL

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("hive-demo") \
    .config("spark.sql.warehouse.dir", "s3a://datalake/warehouse/") \
    .config("hive.metastore.uris", "thrift://hms-host:9083") \
    .enableHiveSupport() \
    .getOrCreate()

# Проверяем подключение
spark.sql("SHOW DATABASES").show()

-- MANAGED таблица: нет EXTERNAL, нет LOCATION
CREATE TABLE silver.transactions (
    txn_id      BIGINT,
    user_id     BIGINT,
    amount      DECIMAL(12,2),
    txn_date    DATE,
    status      STRING
)
USING PARQUET
PARTITIONED BY (txn_date);
-- Данные будут созданы в: s3a://datalake/warehouse/silver.db/transactions/

DROP TABLE silver.transactions;
-- Шаг 1: HMS удаляет запись из TBLS, COLUMNS_V2, PARTITIONS
-- Шаг 2: Spark инициирует удаление ДИРЕКТОРИИ s3a://datalake/warehouse/silver.db/transactions/
-- Все Parquet-файлы внутри УДАЛЕНЫ ФИЗИЧЕСКИ и безвозвратно

-- EXTERNAL таблица: явный EXTERNAL и LOCATION
CREATE EXTERNAL TABLE silver.transactions (
    txn_id      BIGINT,
    user_id     BIGINT,
    amount      DECIMAL(12,2),
    txn_date    DATE,
    status      STRING
)
USING PARQUET
PARTITIONED BY (txn_date)
LOCATION 's3a://datalake/silver/transactions/';
-- Данные лежат по указанному пути, HMS только регистрирует путь

DROP TABLE silver.transactions;
-- Шаг 1: HMS удаляет запись из TBLS, COLUMNS_V2, PARTITIONS
-- Шаг 2: НИЧЕГО БОЛЬШЕ. Файлы в s3a://datalake/silver/transactions/ НЕ ТРОНУТЫ

-- Данные не потеряны - заново регистрируем
CREATE EXTERNAL TABLE silver.transactions (...)
USING PARQUET
LOCATION 's3a://datalake/silver/transactions/';
-- Voilà: таблица восстановлена, все данные доступны

-- Hive DDL style (работает с enableHiveSupport):
CREATE EXTERNAL TABLE silver.orders (...) STORED AS PARQUET LOCATION '...';

-- Spark SQL style (рекомендуется):
CREATE TABLE silver.orders (...) USING PARQUET LOCATION '...';
-- Наличие LOCATION делает таблицу External автоматически!

-- Явно EXTERNAL:
CREATE EXTERNAL TABLE silver.orders (...) USING PARQUET LOCATION '...';

DESCRIBE FORMATTED silver.orders;
-- ...
-- Type: EXTERNAL   ← если есть LOCATION
-- Type: MANAGED    ← если LOCATION нет

s3a://datalake/silver/orders/
  order_date=2024-01-01/
    part-00001-abc.parquet
    part-00002-abc.parquet
  order_date=2024-01-02/
    part-00001-def.parquet
  ...
  order_date=2024-12-31/
    part-00001-xyz.parquet

-- Одна колонка партиционирования
CREATE EXTERNAL TABLE silver.orders (
    order_id    BIGINT,
    customer_id BIGINT,
    total       DECIMAL(12,2),
    status      STRING
    -- order_date НЕТ в основной схеме! Она вынесена в PARTITIONED BY
)
USING PARQUET
PARTITIONED BY (order_date DATE)
LOCATION 's3a://datalake/silver/orders/';

-- Несколько колонок партиционирования (иерархия директорий)
CREATE EXTERNAL TABLE silver.events (
    event_id    STRING,
    user_id     BIGINT,
    event_type  STRING,
    payload     STRING
)
USING PARQUET
PARTITIONED BY (event_date DATE, country STRING)
LOCATION 's3a://datalake/silver/events/';
-- Структура: event_date=2024-01-15/country=RU/part-0001.parquet

s3a://datalake/silver/events/
  event_date=2024-01-01/
    country=RU/
      part-0001.parquet   ← строки за 2024-01-01, страна RU
      part-0002.parquet
    country=US/
      part-0001.parquet   ← строки за 2024-01-01, страна US
  event_date=2024-01-02/
    country=RU/
      part-0001.parquet
    country=DE/
      part-0001.parquet

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, current_timestamp

spark = SparkSession.builder \
    .appName("partitions-demo") \
    .enableHiveSupport() \
    .getOrCreate()

# Допустим, orders_df содержит колонку order_date
orders_df = spark.read.format("delta").load("s3a://datalake/bronze/orders/")

# Spark автоматически создаст директории order_date=.../
orders_df.write \
    .format("parquet") \
    .mode("overwrite") \
    .partitionBy("order_date") \
    .option("path", "s3a://datalake/silver/orders/") \
    .saveAsTable("silver.orders")
    # Spark: создаёт директории, регистрирует партиции в HMS

new_orders = spark.read.format("delta") \
    .load("s3a://datalake/bronze/orders/") \
    .filter(col("order_date") == "2024-01-15")

new_orders.write \
    .format("parquet") \
    .mode("append") \
    .partitionBy("order_date") \
    .option("path", "s3a://datalake/silver/orders/") \
    .saveAsTable("silver.orders")
    # Добавляет только новую партицию order_date=2024-01-15

-- Динамический INSERT: Spark сам определяет значения партиций из данных
INSERT INTO silver.orders PARTITION (order_date)
SELECT order_id, customer_id, total, status, order_date
FROM bronze.raw_orders
WHERE order_date = '2024-01-15';

-- INSERT OVERWRITE: перезаписывает конкретную партицию
INSERT OVERWRITE silver.orders PARTITION (order_date = '2024-01-15')
SELECT order_id, customer_id, total, status
FROM bronze.raw_orders
WHERE order_date = '2024-01-15';
-- ВАЖНО: при OVERWRITE указываем статическое значение → только эта партиция перезаписывается
-- Остальные партиции НЕ ТРОНУТЫ

# Режим STATIC (по умолчанию): перезаписывает ВСЕ партиции
# Опасен: если вставляете только новый день - удалите все остальные дни!
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "static")

# Режим DYNAMIC: перезаписывает только затронутые партиции
# Рекомендуется для инкрементальных загрузок
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

orders_df.write \
    .format("parquet") \
    .mode("overwrite") \
    .partitionBy("order_date") \
    .save("s3a://datalake/silver/orders/")
# С DYNAMIC: только партиции из orders_df будут перезаписаны
# Партиции отсутствующих дней - не тронуты

# Ситуация: данные загружены напрямую в S3
s3a://datalake/silver/orders/
  order_date=2024-01-14/   ← HMS знает (Spark писал)
    part-0001.parquet
  order_date=2024-01-15/   ← HMS НЕ знает (загружено напрямую)
    part-0001.parquet       ← файл есть, но Spark его не видит!

SELECT COUNT(*) FROM silver.orders WHERE order_date = '2024-01-15';
-- Результат: 0 (ноль! хотя данные физически есть)

MSCK REPAIR TABLE silver.orders;
-- Вывод:
-- Partitions not in metastore: silver.orders:order_date=2024-01-15
-- Repair: 1 partition(s) are recovered.

SHOW PARTITIONS silver.orders;
-- order_date=2024-01-14
-- order_date=2024-01-15   ← теперь видна

-- Регистрируем конкретную партицию
ALTER TABLE silver.orders ADD PARTITION (order_date = '2024-01-15');
-- HMS записывает: order_date=2024-01-15 → s3a://datalake/silver/orders/order_date=2024-01-15/

-- Несколько партиций за раз
ALTER TABLE silver.orders
ADD PARTITION (order_date = '2024-01-15')
ADD PARTITION (order_date = '2024-01-16')
ADD PARTITION (order_date = '2024-01-17');

-- С кастомным путём (если директория не совпадает с Hive naming convention)
ALTER TABLE silver.orders
ADD PARTITION (order_date = '2024-01-15')
LOCATION 's3a://datalake/silver/orders/custom-path-for-jan-15/';
-- Парт. имеет свой path - отличается от стандартного key=value/

-- Если партиция уже есть - IF NOT EXISTS защищает от ошибки
ALTER TABLE silver.orders ADD IF NOT EXISTS PARTITION (order_date = '2024-01-15');

-- Удалить партицию
ALTER TABLE silver.orders DROP PARTITION (order_date = '2024-01-01');
-- Для EXTERNAL таблицы: удаляется только запись в HMS, файлы на S3 остаются
-- Для MANAGED таблицы: удаляется запись в HMS И физические файлы

-- С условием (диапазон)
ALTER TABLE silver.orders DROP PARTITION (order_date < '2023-01-01');
-- Удаляет все партиции до 2023 года

-- Предохранитель: IF EXISTS
ALTER TABLE silver.orders DROP IF EXISTS PARTITION (order_date = '2024-01-01');

-- Все партиции таблицы
SHOW PARTITIONS silver.orders;
-- order_date=2024-01-01
-- order_date=2024-01-02
-- ...

-- Фильтрация (только конкретное значение)
SHOW PARTITIONS silver.events PARTITION (event_date = '2024-01-15');
-- event_date=2024-01-15/country=DE
-- event_date=2024-01-15/country=RU
-- event_date=2024-01-15/country=US

-- Количество партиций (через count)
spark.sql("SHOW PARTITIONS silver.orders").count()

from datetime import date, timedelta
from pyspark.sql.functions import col, current_timestamp

def load_partition(spark, partition_date: date) -> None:
    """Загружает одну партицию инкрементально."""
    date_str = partition_date.isoformat()

    # 1. Читаем данные за дату из Bronze
    daily_data = spark.read.format("delta") \
        .load("s3a://datalake/bronze/orders/") \
        .filter(col("order_date") == date_str)

    if daily_data.isEmpty():
        print(f"No data for {date_str}, skipping")
        return

    # 2. Трансформируем
    silver_data = daily_data.select(
        col("order_id"),
        col("customer_id"),
        col("total_amount"),
        col("status"),
        current_timestamp().alias("_ingested_at"),
    )

    # 3. Пишем в S3
    output_path = f"s3a://datalake/silver/orders/order_date={date_str}/"
    silver_data.write \
        .format("parquet") \
        .mode("overwrite") \
        .save(output_path)

    # 4. Регистрируем партицию в HMS точечно (не MSCK!)
    spark.sql(f"""
        ALTER TABLE silver.orders ADD IF NOT EXISTS
        PARTITION (order_date = '{date_str}')
        LOCATION '{output_path}'
    """)
    print(f"Loaded partition: order_date={date_str}")


# Загружаем последние 3 дня
today = date.today()
for i in range(3):
    load_partition(spark, today - timedelta(days=i))

result = spark.sql("""
    SELECT SUM(total_amount)
    FROM silver.orders
    WHERE order_date BETWEEN '2024-01-01' AND '2024-01-31'
""")
result.explain(mode="formatted")

FileScan parquet silver.orders [order_id, total_amount, order_date]
  PartitionFilters: [isnotnull(order_date#10),
                     (order_date#10 >= 2024-01-01),
                     (order_date#10 <= 2024-01-31)]
  PushedFilters: []
  ReadSchema: struct<order_id:bigint, total_amount:decimal(12,2)>
  Location: s3a://datalake/silver/orders/
  SelectedFiles: 31 files from 31 partitions  ← только январь!

-- Pruning РАБОТАЕТ: order_date - partition key
SELECT * FROM silver.orders WHERE order_date = '2024-01-15';

-- Pruning НЕ РАБОТАЕТ: customer_id - обычная колонка, не partition key
SELECT * FROM silver.orders WHERE customer_id = 12345;
-- Spark прочитает ВСЕ файлы и потом отфильтрует строки

-- Pruning ЧАСТИЧНО работает: один из двух фильтров на partition key
SELECT * FROM silver.orders
WHERE order_date = '2024-01-15' AND customer_id = 12345;
-- Spark прочитает только order_date=2024-01-15/, потом фильтрует customer_id

-- КАТАСТРОФА: партиционирование по user_id
CREATE EXTERNAL TABLE silver.events (
    event_id STRING,
    event_type STRING,
    amount DECIMAL
)
USING PARQUET
PARTITIONED BY (user_id BIGINT)  -- ← 50 миллионов уникальных пользователей!
LOCATION 's3a://datalake/silver/events/';

-- ПЛОХО: партиционирование по минутам (high cardinality по времени)
PARTITIONED BY (event_minute STRING)
-- event_minute=2024-01-15T10:30/ → один файл за минуту → миллион файлов в год

-- ХОРОШО: партиционирование по дате
PARTITIONED BY (event_date DATE)
-- event_date=2024-01-15/ → разумное количество файлов за день

-- ПЛОХО: таблица 10 TB без партиционирования
CREATE EXTERNAL TABLE silver.events (
    event_id STRING, event_date DATE, user_id BIGINT, ...
)
USING PARQUET LOCATION 's3a://datalake/silver/events/';
-- Любой запрос с WHERE event_date = '...' → скан 10 TB

-- ИЗБЫТОЧНО: 4 уровня партиционирования
PARTITIONED BY (year INT, month INT, day INT, hour INT)
-- year=2024/month=01/day=15/hour=10/ → 8760 директорий в год
-- Catalyst при запросе без всех 4 фильтров сканирует слишком много метаданных

-- ЛУЧШЕ: один или два уровня
PARTITIONED BY (event_date DATE)
-- Плюс partition pruning по диапазону дат работает отлично
-- Если нужно hourly: event_date=2024-01-15/ и крупные файлы внутри

-- Добавить новую колонку
ALTER TABLE silver.orders ADD COLUMNS (promo_code STRING);

-- Существующие файлы: promo_code = null при чтении
-- Новые файлы: promo_code содержит значение

-- Важно: нельзя добавить колонку в PARTITIONED BY через ALTER TABLE
-- Новый partition key - только через создание новой таблицы

-- Безопасно: DATE → STRING (при чтении Hive-style партиций ключ всегда строка)
ALTER TABLE silver.orders CHANGE COLUMN order_date order_date STRING;

-- ОПАСНО: изменение типа затрагивает имена директорий!
-- Директории event_date=2024-01-15 уже созданы как DATE-формат
-- Смена на другой формат может сломать partition resolution

-- Delta таблица регистрируется в HMS как "delta" provider
CREATE EXTERNAL TABLE silver.orders_delta (
    order_id BIGINT, customer_id BIGINT, total DECIMAL(12,2)
)
USING DELTA
PARTITIONED BY (order_date DATE)
LOCATION 's3a://datalake/silver/orders_delta/';

-- Для Delta: используйте DESCRIBE DETAIL вместо SHOW PARTITIONS
DESCRIBE DETAIL silver.orders_delta;
-- Покажет numFiles, numPartitions, sizeInBytes

CREATE EXTERNAL TABLE silver.orders_iceberg (
    order_id BIGINT, customer_id BIGINT, total DECIMAL(12,2), order_date DATE
)
USING ICEBERG
LOCATION 's3a://datalake/silver/orders_iceberg/';

spark = SparkSession.builder \
    .appName("hive-practice") \
    .config("spark.sql.warehouse.dir", "/tmp/spark-warehouse") \
    .enableHiveSupport() \
    .getOrCreate()

# Создаём MANAGED таблицу
spark.sql("""
    CREATE TABLE demo_managed (
        id BIGINT, name STRING, amount DECIMAL(10,2)
    )
    USING PARQUET
""")

# Заполняем данными
spark.sql("""
    INSERT INTO demo_managed VALUES
    (1, 'Alice', 1500.00),
    (2, 'Bob',   2200.00),
    (3, 'Carol',  800.00)
""")

spark.sql("SELECT * FROM demo_managed").show()

# Проверяем где физически лежат файлы
detail = spark.sql("DESCRIBE FORMATTED demo_managed").collect()
for row in detail:
    if row["col_name"] in ["Location", "Type"]:
        print(f"{row['col_name']}: {row['data_type']}")
# Type:     MANAGED
# Location: /tmp/spark-warehouse/demo_managed

# КАТАСТРОФА: DROP TABLE удаляет данные физически
spark.sql("DROP TABLE demo_managed")
# Файл /tmp/spark-warehouse/demo_managed/ УДАЛЁН
# Восстановить невозможно!
print("Данные безвозвратно удалены!")

import os
from pyspark.sql import Row
from pyspark.sql.types import *
from pyspark.sql.functions import col, current_date, lit, to_date
from datetime import date, timedelta
import random

# Создаём базу данных
spark.sql("CREATE DATABASE IF NOT EXISTS silver_demo LOCATION '/tmp/silver_demo'")

# Схема данных
order_schema = StructType([
    StructField("order_id", LongType()),
    StructField("customer_id", LongType()),
    StructField("total_amount", DoubleType()),
    StructField("status", StringType()),
])

# Создаём EXTERNAL партиционированную таблицу
spark.sql("""
    CREATE EXTERNAL TABLE IF NOT EXISTS silver_demo.orders (
        order_id     BIGINT  COMMENT 'Unique order identifier',
        customer_id  BIGINT  COMMENT 'FK to customers',
        total_amount DOUBLE  COMMENT 'Order total in RUB',
        status       STRING  COMMENT 'completed/pending/cancelled'
    )
    USING PARQUET
    PARTITIONED BY (order_date DATE)
    LOCATION '/tmp/silver_demo_orders/'
    COMMENT 'Silver: normalized orders partitioned by date'
    TBLPROPERTIES ('pipeline.owner' = 'data-team')
""")

# Проверяем тип: должно быть EXTERNAL
spark.sql("DESCRIBE FORMATTED silver_demo.orders") \
    .filter(col("col_name") == "Type") \
    .show(truncate=False)

# Генерируем данные за 3 дня
all_data = []
for day_offset in range(3):
    order_date = date(2024, 1, 15) + timedelta(days=day_offset)
    for i in range(100):
        all_data.append(Row(
            order_id=day_offset * 1000 + i,
            customer_id=random.randint(1, 500),
            total_amount=round(random.uniform(100, 5000), 2),
            status=random.choice(["completed", "pending", "cancelled"]),
            order_date=order_date,
        ))

schema_with_date = StructType([
    StructField("order_id", LongType()),
    StructField("customer_id", LongType()),
    StructField("total_amount", DoubleType()),
    StructField("status", StringType()),
    StructField("order_date", DateType()),
])

orders_df = spark.createDataFrame(all_data, schema_with_date)

# Spark записывает и автоматически регистрирует партиции в HMS
orders_df.write \
    .format("parquet") \
    .mode("append") \
    .partitionBy("order_date") \
    .option("path", "/tmp/silver_demo_orders/") \
    .saveAsTable("silver_demo.orders")

# Проверяем партиции
spark.sql("SHOW PARTITIONS silver_demo.orders").show()

+------------------+
|         partition|
+------------------+
|order_date=2024-01-15|
|order_date=2024-01-16|
|order_date=2024-01-17|
+------------------+

import os
import shutil

# Имитируем: данные добавлены напрямую в S3, минуя Spark
new_date = "2024-01-18"
new_path = f"/tmp/silver_demo_orders/order_date={new_date}/"
os.makedirs(new_path, exist_ok=True)

# Создаём "сторонние" данные и кладём как Parquet-файл напрямую
new_orders = spark.createDataFrame(
    [(5000 + i, random.randint(1, 500), round(random.uniform(100, 5000), 2), "completed")
     for i in range(50)],
    StructType([
        StructField("order_id", LongType()),
        StructField("customer_id", LongType()),
        StructField("total_amount", DoubleType()),
        StructField("status", StringType()),
    ])
)
new_orders.write.format("parquet").mode("overwrite").save(new_path)

# Проверяем: HMS не знает об этой партиции
print("SHOW PARTITIONS (до REPAIR):")
spark.sql("SHOW PARTITIONS silver_demo.orders").show()
# Только 3 партиции (2024-01-15, 16, 17) - 2024-01-18 нет!

# Запрос к "невидимой" партиции
count_invisible = spark.sql(
    "SELECT COUNT(*) FROM silver_demo.orders WHERE order_date = '2024-01-18'"
).collect()[0][0]
print(f"Строк за 2024-01-18 (до REPAIR): {count_invisible}")  # → 0

# Исправляем через MSCK REPAIR TABLE
spark.sql("MSCK REPAIR TABLE silver_demo.orders")
print("\nMSCK REPAIR завершён")

# Теперь партиция видна
print("\nSHOW PARTITIONS (после REPAIR):")
spark.sql("SHOW PARTITIONS silver_demo.orders").show()

count_fixed = spark.sql(
    "SELECT COUNT(*) FROM silver_demo.orders WHERE order_date = '2024-01-18'"
).collect()[0][0]
print(f"Строк за 2024-01-18 (после REPAIR): {count_fixed}")  # → 50

# Таблица с данными за 3 дня (300 строк)
# Тест 1: без партиционного фильтра - читает всё
full_scan = spark.sql("SELECT COUNT(*) FROM silver_demo.orders")
full_scan.explain(mode="formatted")
# PartitionFilters: [] ← нет фильтра по партиции

# Тест 2: с партиционным фильтром - читает только один день
filtered = spark.sql("""
    SELECT COUNT(*) FROM silver_demo.orders
    WHERE order_date = '2024-01-15'
""")
filtered.explain(mode="formatted")
# PartitionFilters: [isnotnull(order_date), EqualTo(order_date, 2024-01-15)]
# SelectedFiles: 1 partition only ← только нужная партиция

# Подтверждение: EXPLAIN показывает разный объём чтения

# Вместо MSCK REPAIR - точечная регистрация только новой партиции
new_date_str = "2024-01-19"
new_path_str = f"/tmp/silver_demo_orders/order_date={new_date_str}/"

# Создаём файл напрямую
new_data = spark.createDataFrame(
    [(6000 + i, random.randint(1, 500), round(random.uniform(100, 5000), 2), "completed")
     for i in range(30)],
    ["order_id", "customer_id", "total_amount", "status"]
)
new_data.write.format("parquet").mode("overwrite").save(new_path_str)

# Регистрируем точечно - мгновенная операция, нет сканирования S3!
spark.sql(f"""
    ALTER TABLE silver_demo.orders
    ADD IF NOT EXISTS PARTITION (order_date = '{new_date_str}')
    LOCATION '{new_path_str}'
""")
print("Партиция зарегистрирована мгновенно через ALTER TABLE ADD PARTITION")

spark.sql("SHOW PARTITIONS silver_demo.orders").show()

# Сохраняем location для проверки
location = spark.sql(
    "DESCRIBE FORMATTED silver_demo.orders"
).filter(col("col_name") == "Location") \
 .collect()[0]["data_type"]

print(f"Данные хранятся в: {location}")

# DROP TABLE на EXTERNAL таблице - безопасно
spark.sql("DROP TABLE silver_demo.orders")
print("DROP TABLE выполнен")

# Проверяем: данные остались!
import os
files_exist = os.path.exists(location)
print(f"Файлы остались на диске: {files_exist}")  # → True

# Восстанавливаем регистрацию
spark.sql(f"""
    CREATE EXTERNAL TABLE silver_demo.orders (
        order_id BIGINT, customer_id BIGINT,
        total_amount DOUBLE, status STRING
    )
    USING PARQUET
    PARTITIONED BY (order_date DATE)
    LOCATION '{location}'
""")
spark.sql("MSCK REPAIR TABLE silver_demo.orders")
print("Таблица восстановлена!")
spark.sql("SELECT COUNT(*) FROM silver_demo.orders").show()

# Список всех баз данных
for db in spark.catalog.listDatabases():
    print(f"DB: {db.name} @ {db.locationUri}")

# Список таблиц с их типами
for t in spark.catalog.listTables("silver_demo"):
    print(f"  {t.name}: type={t.tableType}, temp={t.isTemporary}")
    # tableType: MANAGED, EXTERNAL, VIEW

# Колонки таблицы (включая partition keys)
for col in spark.catalog.listColumns("silver_demo", "orders"):
    print(f"  {col.name}: {col.dataType} (part={col.isPartition})")

# Проверка существования
print(spark.catalog.tableExists("silver_demo.orders"))

# Refresh после внешних изменений
spark.catalog.refreshTable("silver_demo.orders")

# Аналог MSCK REPAIR через API (Spark 3.x)
spark.sql("MSCK REPAIR TABLE silver_demo.orders")

DESCRIBE FORMATTED silver_demo.orders;

# Detailed Table Information
Database:            silver_demo
Table:               orders
Owner:               spark
Type:                EXTERNAL    ← тип таблицы
Provider:            parquet
Location:            /tmp/silver_demo_orders
Comment:             Silver: normalized orders

# Partition Information
# col_name    data_type
order_date    date

# Storage Information
SerDe Library:       org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe
InputFormat:         org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat
OutputFormat:        org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat
Compressed:          No
Num Buckets:         -1

# Table Properties
pipeline.owner       data-team

def hms_table_audit(spark, full_table_name: str) -> dict:
    """Сводная информация о Hive Metastore таблице."""
    parts = full_table_name.split(".")
    db, table = parts[0], parts[1]

    # Основные метаданные
    meta_rows = spark.sql(f"DESCRIBE FORMATTED {full_table_name}").collect()
    meta = {row["col_name"].strip(): row["data_type"].strip()
            for row in meta_rows if row["col_name"].strip()}

    # Количество партиций
    try:
        num_partitions = spark.sql(f"SHOW PARTITIONS {full_table_name}").count()
    except Exception:
        num_partitions = 0  # таблица не партиционирована

    return {
        "table": full_table_name,
        "type": meta.get("Type", "unknown"),
        "location": meta.get("Location", "unknown"),
        "provider": meta.get("Provider", "unknown"),
        "num_partitions": num_partitions,
        "comment": meta.get("Comment", ""),
    }


audit = hms_table_audit(spark, "silver_demo.orders")
for k, v in audit.items():
    print(f"  {k:20s}: {v}")

Production Hive Metastore Table Checklist:
  [ ] Всегда EXTERNAL с явным LOCATION - никогда MANAGED в production
  [ ] Формат: USING PARQUET или USING DELTA (не оставлять Hive SerDe)
  [ ] Партиционирование по DATE или по DATE+CATEGORY (не по ID/timestamp)
  [ ] Cardinality partition-ключа < 50 000 уникальных значений
  [ ] COMMENT на таблице и на каждой колонке
  [ ] TBLPROPERTIES: pipeline.owner, pipeline.version, data.retention_days
  [ ] IF NOT EXISTS для идемпотентности DDL
  [ ] ALTER TABLE ADD PARTITION вместо MSCK REPAIR для incremental pipelines
  [ ] MSCK REPAIR только для одноразовой синхронизации, не в регулярных джобах

MSCK REPAIR TABLE:
  ✓ Одноразовая синхронизация после bulk-загрузки данных
  ✓ Начальная регистрация существующих данных в HMS
  ✓ Таблица имеет < 1000 партиций
  ✗ Не использовать в регулярных ETL-пайплайнах
  ✗ Не использовать на таблицах с 10K+ партиций (слишком медленно)

ALTER TABLE ADD PARTITION:
  ✓ Incremental ETL: добавление новых дней/партиций
  ✓ Высокочастотные пайплайны (hourly, daily)
  ✓ Когда известен точный путь и значение партиции
  ✓ Мгновенная операция - нет сканирования S3