Catalog API: CREATE TABLE, ALTER TABLE, DESCRIBE DETAIL

from pyspark.sql import SparkSession

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

# Проверяем текущий каталог
print(spark.catalog.currentDatabase())   # → default
print(spark.catalog.listDatabases())     # → все базы данных

catalog_name.database_name.table_name
│              │              │
│              │              └── Имя таблицы: orders
│              └── База данных/схема: gold
└── Каталог: spark_catalog (default) или hive_metastore

-- Managed таблица
CREATE TABLE gold.orders (
    order_id LONG,
    customer_id LONG,
    total_amount DOUBLE,
    order_date DATE
) USING DELTA;
-- Данные будут в: <warehouse.dir>/gold.db/orders/

-- External таблица
CREATE TABLE gold.orders (
    order_id LONG,
    customer_id LONG,
    total_amount DOUBLE,
    order_date DATE
) USING DELTA
LOCATION 's3a://datalake/gold/orders/';

-- Простое создание
CREATE DATABASE IF NOT EXISTS silver;

-- С расположением (для external databases)
CREATE DATABASE IF NOT EXISTS gold
LOCATION 's3a://datalake/gold/'
COMMENT 'Gold layer: curated analytical datasets for BI and ML'
WITH DBPROPERTIES (
    'owner' = 'data-platform-team',
    'created_date' = '2024-01-01',
    'sla' = '99.9%'
);

SHOW DATABASES;
SHOW DATABASES LIKE 'gold*';   -- фильтрация по паттерну

DESCRIBE DATABASE EXTENDED gold;
-- Name: gold
-- Location: s3a://datalake/gold/
-- Comment: Gold layer: curated...
-- Properties: ((owner,data-platform-team), ...)

# Переключиться на базу данных
spark.catalog.setCurrentDatabase("gold")

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

CREATE TABLE IF NOT EXISTS silver.order_items (
    item_id         BIGINT      COMMENT 'Surrogate key',
    order_id        BIGINT      NOT NULL COMMENT 'FK to orders',
    sku             STRING      NOT NULL,
    product_name    STRING,
    category        STRING,
    quantity        INT         NOT NULL,
    unit_price      DECIMAL(12,2),
    discount_pct    DECIMAL(5,4),
    line_total      DECIMAL(14,2) COMMENT 'qty * unit_price * (1 - discount_pct)',
    order_date      DATE        NOT NULL COMMENT 'Partition column',
    _ingested_at    TIMESTAMP   COMMENT 'ETL ingestion timestamp',
    _source         STRING      COMMENT 'Source system identifier'
)
USING DELTA
PARTITIONED BY (order_date)
LOCATION 's3a://datalake/silver/order_items/'
COMMENT 'Silver: normalized order line items from Bronze orders JSON'
TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true',
    'pipeline.owner' = 'data-engineering',
    'pipeline.version' = '2.1.0',
    'pipeline.schedule' = '0 */6 * * *',
    'data.retention_days' = '365',
    'data.pii_columns' = 'none',
    'docs.confluence' = 'https://confluence.example.com/display/DE/silver-order-items'
);

-- Delta Lake: ACID, time travel, schema evolution
CREATE TABLE ... USING DELTA ...

-- Apache Iceberg: аналог Delta, vendor-neutral
CREATE TABLE ... USING ICEBERG ...

-- Чистый Parquet: нет ACID, нет time travel
CREATE TABLE ... USING PARQUET ...

-- ORC: исторически популярен в Hive-мире
CREATE TABLE ... USING ORC ...

-- Antиpattern: не указывать USING (Hive SerDe)
CREATE TABLE ... -- ← зависит от spark.sql.sources.default

CREATE TABLE gold.revenue_by_category
USING DELTA
LOCATION 's3a://datalake/gold/revenue_by_category/'
PARTITIONED BY (order_date)
TBLPROPERTIES ('pipeline.type' = 'aggregated')
AS
SELECT
    order_date,
    category,
    SUM(line_total)       AS total_revenue,
    COUNT(DISTINCT order_id) AS order_count,
    SUM(quantity)         AS units_sold,
    AVG(unit_price)       AS avg_unit_price
FROM silver.order_items
WHERE order_date >= '2024-01-01'
GROUP BY order_date, category;

# То же самое в PySpark
revenue_df = spark.sql("""
    SELECT order_date, category, SUM(line_total) AS total_revenue
    FROM silver.order_items
    GROUP BY order_date, category
""")

revenue_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("order_date") \
    .saveAsTable("gold.revenue_by_category")

CREATE TABLE silver.users_bucketed
USING PARQUET
CLUSTERED BY (user_id) INTO 64 BUCKETS
LOCATION 's3a://datalake/silver/users_bucketed/'
AS SELECT * FROM silver.users;

# DataFrame → Temporary View
df = spark.read.format("delta").load("s3a://datalake/silver/orders/")
df.createOrReplaceTempView("orders_silver")

# Теперь можно использовать в SQL
result = spark.sql("""
    SELECT customer_id, COUNT(*) AS order_count
    FROM orders_silver
    GROUP BY customer_id
""")

# View живёт только пока жива SparkSession
# spark.stop() → view исчезает

df.createOrReplaceGlobalTempView("shared_reference_data")

# Из другой сессии
spark2.sql("SELECT * FROM global_temp.shared_reference_data")

-- Добавить одну колонку
ALTER TABLE silver.order_items
ADD COLUMN discount_code STRING
COMMENT 'Promo code applied at checkout, nullable';

-- Добавить несколько колонок сразу
ALTER TABLE silver.order_items
ADD COLUMNS (
    campaign_id STRING COMMENT 'Marketing campaign identifier',
    channel     STRING COMMENT 'Acquisition channel: web/mobile/api',
    is_gift     BOOLEAN COMMENT 'Is this a gift order'
);

# Эквивалент в PySpark через SQL
spark.sql("""
    ALTER TABLE silver.order_items
    ADD COLUMN discount_code STRING COMMENT 'Promo code applied at checkout'
""")

# Или через DataFrame API с mergeSchema
df_with_new_col = df.withColumn("discount_code", lit(None).cast("string"))
df_with_new_col.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("silver.order_items")

-- Переименовать колонку (Delta Lake 2.0+ / Iceberg)
ALTER TABLE silver.order_items
RENAME COLUMN discount_pct TO discount_rate;

-- Безопасно: расширение типа
ALTER TABLE silver.order_items
ALTER COLUMN quantity TYPE BIGINT;
-- INT → BIGINT: все существующие значения корректно конвертируются

-- Безопасно: от числа с меньшей точностью к большей
ALTER TABLE silver.order_items
ALTER COLUMN unit_price TYPE DECIMAL(14,2);
-- DECIMAL(12,2) → DECIMAL(14,2): расширение диапазона

-- НЕБЕЗОПАСНО (Delta заблокирует):
-- ALTER COLUMN unit_price TYPE DOUBLE;  -- DECIMAL → DOUBLE потеря точности
-- ALTER COLUMN order_id TYPE INT;       -- BIGINT → INT: потенциальное переполнение
-- ALTER COLUMN order_date TYPE STRING;  -- DATE → STRING: разные семантики

AnalysisException: Cannot change column 'unit_price' from DECIMAL(12,2) to DOUBLE

-- Обновить версию пайплайна и добавить новые свойства
ALTER TABLE silver.order_items
SET TBLPROPERTIES (
    'pipeline.version' = '2.2.0',
    'pipeline.last_updated' = '2024-03-01',
    'data.quality_score' = '0.998',
    'data.freshness_sla' = '1h'
);

-- Удалить свойство
ALTER TABLE silver.order_items
UNSET TBLPROPERTIES ('docs.confluence');

-- Переключить таблицу на новый S3-бакет (данные уже скопированы)
ALTER TABLE gold.revenue_by_category
SET LOCATION 's3a://new-datalake/gold/revenue_by_category/';

-- External таблица: удалится только metadata, данные останутся
DROP TABLE IF EXISTS silver.order_items;
-- Файлы в s3a://datalake/silver/order_items/ остаются нетронутыми

-- Managed таблица: удалятся и metadata, и данные на диске!
DROP TABLE gold.temp_aggregation;
-- ← физические файлы УДАЛЕНЫ

-- TRUNCATE: удаляет данные, но оставляет metadata и схему
TRUNCATE TABLE silver.order_items;
-- Все строки удалены, таблица пуста, но существует в каталоге

DESCRIBE TABLE silver.order_items;

+-------------------+-----------+--------------------------------------------------+
|          col_name |  data_type|                                           comment|
+-------------------+-----------+--------------------------------------------------+
|            item_id|     bigint|                                    Surrogate key|
|           order_id|     bigint|                               FK to orders      |
|                sku|     string|                                                  |
|       product_name|     string|                                                  |
|           category|     string|                                                  |
|           quantity|        int|                                                  |
|         unit_price|decimal(12,2)|                                                |
|       discount_pct|decimal(5,4)|                                                 |
|         line_total|decimal(14,2)|    qty * unit_price * (1 - discount_pct)        |
|         order_date|       date|                              Partition column    |
|      _ingested_at|  timestamp|                         ETL ingestion timestamp   |
|            _source|     string|                      Source system identifier    |
+-------------------+-----------+--------------------------------------------------+

DESCRIBE EXTENDED silver.order_items;
-- или
DESCRIBE FORMATTED silver.order_items;

# Detailed Table Information
Database:           silver
Table:              order_items
Owner:              spark
Created Time:       Mon Jan 15 10:00:00 UTC 2024
Last Access:        UNKNOWN
Created By:         Spark 3.5.0
Type:               EXTERNAL
Provider:           delta
Location:           s3a://datalake/silver/order_items
Serde Library:      org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
InputFormat:        org.apache.hadoop.mapred.SequenceFileInputFormat
OutputFormat:       org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat

# Partition Information
# col_name    data_type    comment
order_date    date         Partition column

# Table Properties
delta.autoOptimize.autoCompact    true
delta.autoOptimize.optimizeWrite  true
delta.minReaderVersion            1
delta.minWriterVersion            2
pipeline.owner                    data-engineering
pipeline.version                  2.1.0

DESCRIBE DETAIL silver.order_items;

+------+-----------------------------+------+------+----------+---------+----------+----------+
|format|              location       |nFiles|nParts|sizeInBytes|minReader|minWriter| properties|
+------+-----------------------------+------+------+----------+---------+----------+----------+
|delta |s3a://datalake/silver/...    |  847 |  365 |4523456789|        1|        2| {delta.au..}|
+------+-----------------------------+------+------+----------+---------+----------+----------+

detail = spark.sql("DESCRIBE DETAIL silver.order_items")
detail.printSchema()

# Извлекаем ключевые метрики
row = detail.collect()[0]
print(f"Format:      {row['format']}")
print(f"Location:    {row['location']}")
print(f"Files:       {row['numFiles']:,}")
print(f"Size:        {row['sizeInBytes'] / 1024**3:.2f} GB")
print(f"Avg file:    {row['sizeInBytes'] / row['numFiles'] / 1024**2:.1f} MB")

# Диагностика мелких файлов
avg_file_mb = row['sizeInBytes'] / row['numFiles'] / 1024**2
if avg_file_mb < 64:
    print(f"⚠ Small file problem: avg file {avg_file_mb:.1f} MB < 64 MB")
    print(f"  Consider running OPTIMIZE on this table")

def audit_table_health(table_name: str) -> None:
    """Аудит здоровья Delta/Iceberg таблицы через DESCRIBE DETAIL."""
    detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]

    size_gb = detail['sizeInBytes'] / 1024**3
    num_files = detail['numFiles']
    avg_file_mb = detail['sizeInBytes'] / max(num_files, 1) / 1024**2

    print(f"\n=== Table Health: {table_name} ===")
    print(f"  Format:          {detail['format']}")
    print(f"  Location:        {detail['location']}")
    print(f"  Total size:      {size_gb:.2f} GB")
    print(f"  Num files:       {num_files:,}")
    print(f"  Avg file size:   {avg_file_mb:.1f} MB")

    # Оценка здоровья файлов
    if avg_file_mb >= 128:
        file_health = "✓ Excellent (>128 MB avg)"
    elif avg_file_mb >= 64:
        file_health = "✓ Good (64-128 MB avg)"
    elif avg_file_mb >= 16:
        file_health = "⚠ Fair (16-64 MB avg) - consider OPTIMIZE"
    else:
        file_health = "✗ Poor (<16 MB avg) - run OPTIMIZE"
    print(f"  File health:     {file_health}")


audit_table_health("silver.order_items")

=== Table Health: silver.order_items ===
  Format:          delta
  Location:        s3a://datalake/silver/order_items
  Total size:      4.21 GB
  Num files:       847
  Avg file size:   5.1 MB
  File health:     ✗ Poor (<16 MB avg) - run OPTIMIZE

-- Все базы данных
SHOW DATABASES;
SHOW DATABASES LIKE 'gold*';

-- Все таблицы в текущей или указанной БД
SHOW TABLES;
SHOW TABLES IN silver;
SHOW TABLES IN silver LIKE 'order*';

-- Партиции таблицы
SHOW PARTITIONS silver.order_items;
-- order_date=2024-01-01
-- order_date=2024-01-02
-- ...

-- Свойства таблицы
SHOW TBLPROPERTIES silver.order_items;
-- pipeline.owner   data-engineering
-- pipeline.version 2.1.0
-- ...

-- Конкретное свойство
SHOW TBLPROPERTIES silver.order_items ('pipeline.version');
-- 2.1.0

-- Колонки (аналог DESCRIBE)
SHOW COLUMNS IN silver.order_items;

# Список баз данных
for db in spark.catalog.listDatabases():
    print(f"Database: {db.name}")
    print(f"  Description: {db.description}")
    print(f"  Location:    {db.locationUri}")
    print()

# Список таблиц в базе данных
for table in spark.catalog.listTables("silver"):
    print(f"  {table.name} ({table.tableType}) - {table.description or 'no description'}")
    # tableType: MANAGED, EXTERNAL, VIEW

# Список колонок
for col in spark.catalog.listColumns("silver", "order_items"):
    nullable = "nullable" if col.nullable else "NOT NULL"
    print(f"  {col.name}: {col.dataType} [{nullable}]")
    if col.description:
        print(f"    → {col.description}")

def ensure_database_exists(db_name: str, location: str, comment: str = "") -> None:
    """Создаёт базу данных если не существует."""
    if not any(db.name == db_name for db in spark.catalog.listDatabases()):
        spark.sql(f"""
            CREATE DATABASE {db_name}
            LOCATION '{location}'
            COMMENT '{comment}'
        """)
        print(f"Created database: {db_name}")
    else:
        print(f"Database already exists: {db_name}")


def table_exists(db_name: str, table_name: str) -> bool:
    """Безопасная проверка существования таблицы."""
    return spark.catalog.tableExists(f"{db_name}.{table_name}")


# Использование в ETL-пайплайне
ensure_database_exists("silver", "s3a://datalake/silver/", "Silver layer")

if not table_exists("silver", "order_items"):
    spark.sql("""
        CREATE TABLE silver.order_items (...)
        USING DELTA
        LOCATION 's3a://datalake/silver/order_items/'
    """)
    print("Created table: silver.order_items")

-- REFRESH TABLE: сбрасывает кэш метаданных Spark
-- Нужно если файлы изменились вне Spark (например, другим инструментом)
REFRESH TABLE silver.order_items;

-- MSCK REPAIR TABLE: обнаруживает новые партиции в external Hive-таблицах
-- Нужно если данные добавлены напрямую в S3 без Spark
MSCK REPAIR TABLE silver.order_items;

# Программное обновление
spark.catalog.refreshTable("silver.order_items")

# Или через SQL
spark.sql("REFRESH TABLE silver.order_items")

# Получить метаданные таблицы программно
table_meta = spark.catalog.getTable("silver", "order_items")
print(f"Name: {table_meta.name}")
print(f"Type: {table_meta.tableType}")  # MANAGED или EXTERNAL
print(f"DB:   {table_meta.database}")

# DESCRIBE DETAIL как DataFrame для программной обработки
detail_df = spark.sql("DESCRIBE DETAIL silver.order_items")
detail = detail_df.collect()[0].asDict()

print(f"Files:     {detail.get('numFiles', 'N/A')}")
print(f"Size:      {detail.get('sizeInBytes', 0) / 1024**3:.2f} GB")
print(f"Partitions:{detail.get('numPartitions', 'N/A')}")

-- Базовая статистика: размер, количество строк
ANALYZE TABLE silver.order_items COMPUTE STATISTICS;

-- Расширенная: дополнительно гистограммы по каждой колонке
ANALYZE TABLE silver.order_items COMPUTE STATISTICS FOR ALL COLUMNS;

-- Только для конкретных колонок
ANALYZE TABLE silver.order_items COMPUTE STATISTICS FOR COLUMNS order_id, category;

# Table Statistics
# Rows: 45,234,891
# Total size: 4,523,456,789 bytes
# Column Statistics:
#   order_id: [min=1, max=45234891, distinct=45234891, null=0]
#   category: [min=accessories, max=toys, distinct=23, null=42]

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, current_timestamp, lit, round as spark_round

spark = SparkSession.builder \
    .appName("catalog-practice") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Создаём базы данных для Bronze, Silver, Gold
for layer in ["bronze", "silver", "gold"]:
    spark.sql(f"""
        CREATE DATABASE IF NOT EXISTS {layer}
        COMMENT '{layer.capitalize()} layer of Medallion architecture'
    """)

spark.sql("SHOW DATABASES").show()

CREATE TABLE IF NOT EXISTS silver.marketing_events (
    event_id        STRING      NOT NULL  COMMENT 'UUID from source system',
    user_id         BIGINT                COMMENT 'User identifier, nullable for anonymous',
    campaign_id     STRING      NOT NULL  COMMENT 'Marketing campaign code',
    channel         STRING      NOT NULL  COMMENT 'Channel: email/push/sms/banner',
    event_type      STRING      NOT NULL  COMMENT 'click/open/conversion/unsubscribe',
    revenue_usd     DECIMAL(10,2)         COMMENT 'Revenue attributed to this event',
    event_date      DATE        NOT NULL  COMMENT 'Partition key',
    _ingested_at    TIMESTAMP             COMMENT 'ETL timestamp'
)
USING DELTA
PARTITIONED BY (event_date)
LOCATION 's3a://datalake/silver/marketing_events/'
COMMENT 'Silver: normalized marketing touchpoint events'
TBLPROPERTIES (
    'pipeline.owner'       = 'marketing-analytics',
    'pipeline.version'     = '1.0.0',
    'pipeline.schedule'    = '0 */4 * * *',
    'data.retention_days'  = '730',
    'delta.autoOptimize.optimizeWrite' = 'true'
);

from pyspark.sql.types import *
import pyspark.sql.functions as F

# Синтетические данные маркетинговых событий
import random
from datetime import date, timedelta

events_data = []
channels = ["email", "push", "sms", "banner"]
event_types = ["click", "open", "conversion", "unsubscribe"]
campaigns = ["SPRING24", "PROMO_Q1", "LOYALTY_GOLD", "REACTIVATION"]

for i in range(10000):
    event_date = date(2024, 1, 1) + timedelta(days=random.randint(0, 89))
    events_data.append((
        f"evt_{i:08d}",
        random.randint(1, 50000) if random.random() > 0.15 else None,
        random.choice(campaigns),
        random.choice(channels),
        random.choice(event_types),
        round(random.uniform(0, 500), 2) if random.random() > 0.7 else None,
        event_date,
        None,  # _ingested_at
    ))

events_schema = StructType([
    StructField("event_id", StringType()),
    StructField("user_id", LongType()),
    StructField("campaign_id", StringType()),
    StructField("channel", StringType()),
    StructField("event_type", StringType()),
    StructField("revenue_usd", DecimalType(10, 2)),
    StructField("event_date", DateType()),
    StructField("_ingested_at", TimestampType()),
])

events_df = spark.createDataFrame(events_data, events_schema) \
    .withColumn("_ingested_at", current_timestamp())

events_df.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("silver.marketing_events")

print(f"Written {events_df.count()} rows")

# Проверяем состояние таблицы после записи
detail = spark.sql("DESCRIBE DETAIL silver.marketing_events").collect()[0]

print("=== Table Audit ===")
print(f"Format:       {detail['format']}")
print(f"Location:     {detail['location']}")
print(f"Num files:    {detail['numFiles']}")
print(f"Size:         {detail['sizeInBytes'] / 1024:.1f} KB")

# Статистика партиций
spark.sql("SHOW PARTITIONS silver.marketing_events").show(10)

# Количество строк по партициям
spark.sql("""
    SELECT event_date, COUNT(*) AS rows
    FROM silver.marketing_events
    GROUP BY event_date
    ORDER BY event_date
    LIMIT 5
""").show()

-- Добавляем колонку для промокода
ALTER TABLE silver.marketing_events
ADD COLUMN discount_code STRING
COMMENT 'Discount code applied, if any. Format: DISC_{TYPE}_{AMOUNT}';

-- Добавляем флаг A/B теста
ALTER TABLE silver.marketing_events
ADD COLUMN ab_variant STRING
COMMENT 'A/B test variant: A, B, or null if not in test';

-- Обновляем версию пайплайна
ALTER TABLE silver.marketing_events
SET TBLPROPERTIES (
    'pipeline.version' = '1.1.0',
    'pipeline.last_schema_change' = '2024-03-01',
    'schema.changelog' = 'v1.1.0: added discount_code, ab_variant'
);

# Новые колонки видны в схеме
spark.sql("DESCRIBE TABLE silver.marketing_events").show(truncate=False)

# Старые строки возвращают null для новых колонок
spark.sql("""
    SELECT event_id, campaign_id, discount_code, ab_variant
    FROM silver.marketing_events
    LIMIT 5
""").show()
# discount_code и ab_variant = null для старых строк - ОК

# Проверяем свойства
spark.sql("SHOW TBLPROPERTIES silver.marketing_events").show(truncate=False)

CREATE TABLE IF NOT EXISTS gold.campaign_performance
USING DELTA
LOCATION 's3a://datalake/gold/campaign_performance/'
PARTITIONED BY (event_date)
COMMENT 'Gold: daily campaign KPIs for BI dashboards'
TBLPROPERTIES (
    'pipeline.source' = 'silver.marketing_events',
    'pipeline.type' = 'aggregated',
    'data.refresh' = 'daily'
)
AS
SELECT
    event_date,
    campaign_id,
    channel,
    COUNT(*)                                          AS total_events,
    COUNT(DISTINCT user_id)                           AS unique_users,
    SUM(CASE WHEN event_type = 'click'        THEN 1 ELSE 0 END) AS clicks,
    SUM(CASE WHEN event_type = 'conversion'   THEN 1 ELSE 0 END) AS conversions,
    SUM(CASE WHEN event_type = 'unsubscribe'  THEN 1 ELSE 0 END) AS unsubscribes,
    ROUND(SUM(revenue_usd), 2)                        AS total_revenue,
    ROUND(SUM(revenue_usd) / NULLIF(COUNT(DISTINCT user_id), 0), 2) AS revenue_per_user
FROM silver.marketing_events
GROUP BY event_date, campaign_id, channel;

DDL_CHECKLIST = """
При создании каждой production-таблицы проверяем:
  [ ] CREATE TABLE IF NOT EXISTS - идемпотентность
  [ ] USING DELTA (или ICEBERG) - явный формат
  [ ] LOCATION - external таблица, данные выживут DROP
  [ ] COMMENT на таблице - что хранит, откуда данные
  [ ] COMMENT на каждой колонке - что означает поле
  [ ] PARTITIONED BY - если > 10 GB и есть временной или категорийный ключ
  [ ] TBLPROPERTIES: pipeline.owner, pipeline.version, data.retention_days
  [ ] NOT NULL на обязательных колонках (Delta / Iceberg поддерживают constraints)
"""
print(DDL_CHECKLIST)

-- Шаблон стандартных TBLPROPERTIES для production
TBLPROPERTIES (
    -- Владение и описание
    'pipeline.owner'             = 'team-name',
    'pipeline.slack_channel'     = '#team-data-alerts',

    -- Версионирование
    'pipeline.version'           = '1.0.0',
    'pipeline.git_commit'        = 'abc1234',
    'pipeline.created_date'      = '2024-01-01',

    -- Расписание
    'pipeline.schedule'          = '0 */6 * * *',
    'pipeline.sla_minutes'       = '30',

    -- Данные
    'data.retention_days'        = '365',
    'data.pii_columns'           = 'email,phone',
    'data.classification'        = 'internal',

    -- Техническое
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact'   = 'true'
)

class TableManager:
    """Управляет жизненным циклом таблицы в Spark Catalog."""

    def __init__(self, spark: SparkSession):
        self.spark = spark

    def create_if_not_exists(self, ddl: str) -> bool:
        """Выполняет CREATE TABLE IF NOT EXISTS и возвращает True если создана."""
        before = set(t.name for t in self.spark.catalog.listTables())
        self.spark.sql(ddl)
        after = set(t.name for t in self.spark.catalog.listTables())
        created = bool(after - before)
        return created

    def get_version(self, table: str) -> str:
        """Читает pipeline.version из TBLPROPERTIES."""
        props = {
            row["key"]: row["value"]
            for row in self.spark.sql(f"SHOW TBLPROPERTIES {table}").collect()
        }
        return props.get("pipeline.version", "unknown")

    def bump_version(self, table: str, new_version: str) -> None:
        """Обновляет версию пайплайна в метаданных."""
        self.spark.sql(f"""
            ALTER TABLE {table}
            SET TBLPROPERTIES ('pipeline.version' = '{new_version}')
        """)
        print(f"Updated {table} version to {new_version}")

    def audit(self, table: str) -> dict:
        """Возвращает сводку здоровья таблицы."""
        detail = self.spark.sql(f"DESCRIBE DETAIL {table}").collect()[0].asDict()
        num_files = detail.get("numFiles", 0)
        size_bytes = detail.get("sizeInBytes", 0)
        avg_mb = size_bytes / max(num_files, 1) / 1024**2
        return {
            "table": table,
            "format": detail.get("format"),
            "size_gb": round(size_bytes / 1024**3, 2),
            "num_files": num_files,
            "avg_file_mb": round(avg_mb, 1),
            "needs_optimize": avg_mb < 64,
        }


# Использование
mgr = TableManager(spark)
audit = mgr.audit("silver.marketing_events")
print(audit)
# {'table': 'silver.marketing_events', 'format': 'delta', 'size_gb': 0.05,
#  'num_files': 12, 'avg_file_mb': 4.3, 'needs_optimize': True}

# НЕПРАВИЛЬНО: вся аналитика через spark.read.parquet(...)
df = spark.read.parquet("s3a://datalake/silver/events/2024/01/15/")
# Проблемы: схема неизвестна, нет документации, нельзя делать SQL,
# путь хардкожен, при переезде данных ломается код

# ПРАВИЛЬНО: зарегистрировать таблицу, работать через каталог
df = spark.table("silver.events")
# или
df = spark.sql("SELECT * FROM silver.events WHERE event_date = '2024-01-15'")

# ОПАСНО: managed таблица (без LOCATION)
spark.sql("CREATE TABLE silver.orders (...) USING DELTA")
# DROP TABLE silver.orders → ДАННЫЕ УДАЛЕНЫ с диска!

# БЕЗОПАСНО: external таблица
spark.sql("""
    CREATE TABLE silver.orders (...)
    USING DELTA
    LOCATION 's3a://datalake/silver/orders/'
""")
# DROP TABLE silver.orders → только metadata, данные живут

# СЛОМАЕТСЯ при повторном запуске пайплайна:
spark.sql("CREATE TABLE silver.events (...)")
# AnalysisException: Table `silver`.`events` already exists

# ИДЕМПОТЕНТНО:
spark.sql("CREATE TABLE IF NOT EXISTS silver.events (...)")

-- ПЛОХО: таблица-загадка
CREATE TABLE analytics.results USING PARQUET AS SELECT ...;
-- Что хранит? Кто создал? Откуда данные? Когда обновляется?

-- ХОРОШО: самодокументируемая таблица
CREATE TABLE analytics.results
USING DELTA
LOCATION '...'
COMMENT 'Daily campaign KPIs aggregated from silver.events. Source: ETL v2.1'
TBLPROPERTIES ('pipeline.owner' = 'analytics-team', 'pipeline.version' = '2.1.0')
AS SELECT ...;