S2: Parquet sin tabla permanente
  → S3: Parquet en HDFS y S3-compatible
  → S4: Spark como motor
  → S5: organización refinada y particionada
  → S6: Hive Metastore y Trino comparten la tabla
  → S7: Iceberg añade gestión de tablas y snapshots

%pip install -q "jupyter-notebook-slide @ git+https://github.com/dsevilla/jupyter-notebook-slide.git"
%load_ext notebook_slide

import notebook_slide as jnbs

# Colores de 26-27/teoria/tcdm.css, adaptados al tema de las sesiones.
jnbs.configure(
    background="#eaf2f8",
    foreground="#1f2933",
    border="#c9d6e1",
    heading="#0c304d",
    subheading="#0c304d",
    link="#174f7a",
    code_background="#eaf2f8",
    code_foreground="#0c304d",
    quote_background="#ffffff",
    font_family="Atkinson Hyperlegible, Inter, Aptos, Segoe UI, Helvetica, Arial, sans-serif",
    font_url="https://fonts.googleapis.com/css2?family=Atkinson+Hyperlegible:ital,wght@0,400;0,700;1,400;1,700&display=swap",
)
jnbs.register_slide_type("avance", "A continuación", "#174f7a")
jnbs.register_slide_type("resumen", "Recapitulación", "#a54467")
jnbs.register_slide_type("pregunta", "Pregunta guía", "#0c304d")
jnbs.register_slide_type("evaluacion", "Evaluación de la sesión", "#b3701a")
jnbs.register_slide_type(
    "titulo",
    "Sección",
    "#174f7a",
    layout="title",
    background="linear-gradient(135deg, #0c304d, #174f7a 62%, #a54467)",
    foreground="#ffffff",
    border="transparent",
    heading="#ffffff",
    subheading="#dceaf4",
)
jnbs.register_alias("diapositiva")

cd ~/tcdm-public
git pull
make -C entorno warehouse-up

!hdfs dfs -ls -d /datalake/raw/tpcds/date_dim \
    /datalake/silver/tpcds/web_sales \
    /datalake/gold/tpcds/web_sales_by_year \
    /warehouse

%%writefile requirements.txt
# Dependencias del kernel interactivo de la sesión 6 (Spark local + cliente
# Trino). La serie de PySpark aceptada por el curso es >4,<4.2, igual que en
# S4 y S5.
pyspark>4,<4.2
# Cliente DB-API 2.0 de Trino: es como este notebook consulta Trino sin usar
# el binario `trino` ni `docker exec` (el kernel corre dentro de namenode, no
# tiene acceso al Docker del host). Se fija la serie mayor para que el tipado
# de columnas y la representación de valores no cambien entre ejecuciones.
trino>=0.330,<1
pandas

%pip install -q -r requirements.txt

import sys

import pandas
import pyspark
import trino

print(f"PySpark {pyspark.__version__} con Python {sys.version.split()[0]}.")
print(f"Cliente trino: {trino.__version__}")
print(f"Pandas: {pandas.__version__}")

from pyspark import SparkContext
from pyspark.sql import SparkSession

HIVE_METASTORE_URIS: str = "thrift://hive-metastore:9083"
WAREHOUSE_DIR: str = "hdfs://namenode:9000/warehouse"

spark: SparkSession = (
    SparkSession.builder.appName("TCDM_S6_Local")
    .master("local[2]")
    .config("spark.driver.memory", "1g")
    .config("spark.sql.shuffle.partitions", "8")
    .config("spark.hadoop.hive.metastore.uris", HIVE_METASTORE_URIS)
    .config("spark.sql.warehouse.dir", WAREHOUSE_DIR)
    .config("spark.sql.catalogImplementation", "hive")
    .enableHiveSupport()
    .getOrCreate()
)
sc: SparkContext = spark.sparkContext

print(f"Spark version: {spark.version} | Master: {sc.master}")
print(f"Metastore URIs: {HIVE_METASTORE_URIS}")
print(f"Warehouse dir: {WAREHOUSE_DIR}")

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

import pandas as pd
from trino.dbapi import Connection, Cursor

trino_conn: Connection = trino.dbapi.connect(
    host="trino-hdfs",
    port=8080,
    user="luser",
    catalog="hive",
    schema="tcdm",
)


def run_trino(sql: str) -> pd.DataFrame:
    '''Ejecuta una sentencia en Trino y devuelve el resultado como DataFrame.'''
    cursor: Cursor = trino_conn.cursor()
    cursor.execute(sql)
    rows: list[tuple] = cursor.fetchall()
    columns: list[str] = [description[0] for description in cursor.description or []]
    return pd.DataFrame(rows, columns=columns)

run_trino("SELECT 1 AS ping")

/datalake/raw/tpcds/date_dim
    → tabla Hive externa: tcdm.date_dim

/datalake/silver/tpcds/web_sales
    → tabla Hive externa particionada: tcdm.web_sales_silver

/datalake/gold/tpcds/web_sales_by_year
    → tabla Hive externa orientada al consumo: tcdm.web_sales_by_year_gold

run_trino("CREATE SCHEMA IF NOT EXISTS hive.tcdm")

run_trino("DROP TABLE IF EXISTS hive.tcdm.date_dim")

create_date_dim_sql: str = '''
CREATE TABLE hive.tcdm.date_dim (
    d_date_sk BIGINT,
    d_date_id CHAR(16),
    d_date DATE,
    d_month_seq INTEGER,
    d_week_seq INTEGER,
    d_quarter_seq INTEGER,
    d_year INTEGER,
    d_dow INTEGER,
    d_moy INTEGER,
    d_dom INTEGER,
    d_qoy INTEGER,
    d_fy_year INTEGER,
    d_fy_quarter_seq INTEGER,
    d_fy_week_seq INTEGER,
    d_day_name CHAR(9),
    d_quarter_name CHAR(6),
    d_holiday CHAR(1),
    d_weekend CHAR(1),
    d_following_holiday CHAR(1),
    d_first_dom INTEGER,
    d_last_dom INTEGER,
    d_same_day_ly INTEGER,
    d_same_day_lq INTEGER,
    d_current_day CHAR(1),
    d_current_week CHAR(1),
    d_current_month CHAR(1),
    d_current_quarter CHAR(1),
    d_current_year CHAR(1)
) WITH (
    format = 'PARQUET',
    external_location = 'hdfs://namenode:9000/datalake/raw/tpcds/date_dim'
)
'''
run_trino(create_date_dim_sql)
print("tcdm.date_dim creada.")

run_trino("SHOW CREATE TABLE hive.tcdm.date_dim")

run_trino("DESCRIBE hive.tcdm.date_dim")

date_dim_count: pd.DataFrame = run_trino("SELECT count(*) AS filas FROM hive.tcdm.date_dim")
print(date_dim_count)

assert int(date_dim_count.loc[0, "filas"]) == 73_049

run_trino('''
    COMMENT ON TABLE hive.tcdm.date_dim IS
    'Dimensión de fechas de TPC-DS SF1, registrada como tabla externa sobre /datalake/raw/tpcds/date_dim (ver S2).'
''')
run_trino("SHOW CREATE TABLE hive.tcdm.date_dim")

run_trino("DROP TABLE IF EXISTS hive.tcdm.web_sales_silver")

create_web_sales_silver_sql: str = '''
CREATE TABLE hive.tcdm.web_sales_silver (
    ws_order_number BIGINT,
    ws_item_sk BIGINT,
    ws_bill_customer_sk BIGINT,
    ws_bill_addr_sk BIGINT,
    ws_sold_date_sk BIGINT,
    ws_sold_date DATE,
    ws_quantity INTEGER,
    ws_list_price DECIMAL(7, 2),
    ws_ext_discount_amt DECIMAL(7, 2),
    ws_ext_sales_price DECIMAL(7, 2),
    ws_net_paid DECIMAL(7, 2),
    ws_net_profit DECIMAL(7, 2),
    sold_year INTEGER,
    sold_month INTEGER
) WITH (
    format = 'PARQUET',
    external_location = 'hdfs://namenode:9000/datalake/silver/tpcds/web_sales',
    partitioned_by = ARRAY['sold_year', 'sold_month']
)
'''
run_trino(create_web_sales_silver_sql)
print("tcdm.web_sales_silver creada.")

run_trino("SHOW CREATE TABLE hive.tcdm.web_sales_silver")

before_repair: pd.DataFrame = run_trino("SELECT count(*) AS filas FROM hive.tcdm.web_sales_silver")
print(before_repair)

sync_partitions_sql: str = '''
CALL hive.system.sync_partition_metadata(
    schema_name => 'tcdm',
    table_name => 'web_sales_silver',
    mode => 'FULL')
'''
run_trino(sync_partitions_sql)
print("Particiones sincronizadas.")

after_repair: pd.DataFrame = run_trino("SELECT count(*) AS filas FROM hive.tcdm.web_sales_silver")
print(after_repair)

assert int(after_repair.loc[0, "filas"]) > 0
assert int(before_repair.loc[0, "filas"]) == 0

partitions_df: pd.DataFrame = run_trino(
    'SELECT * FROM hive.tcdm."web_sales_silver$partitions" ORDER BY sold_year, sold_month'
)
print(f"Particiones registradas: {len(partitions_df)}")
partitions_df

explain_unfiltered: pd.DataFrame = run_trino(
    "EXPLAIN SELECT count(*) FROM hive.tcdm.web_sales_silver"
)
print("\n".join(explain_unfiltered.iloc[:, 0].tolist()))

explain_filtered: pd.DataFrame = run_trino(
    '''EXPLAIN SELECT count(*) FROM hive.tcdm.web_sales_silver
    WHERE sold_year = 2000 AND sold_month = 1'''
)
print("\n".join(explain_filtered.iloc[:, 0].tolist()))

explain_analyze_filtered: pd.DataFrame = run_trino(
    '''EXPLAIN ANALYZE SELECT count(*) FROM hive.tcdm.web_sales_silver
    WHERE sold_year = 2000 AND sold_month = 1'''
)
print("\n".join(explain_analyze_filtered.iloc[:, 0].tolist()))

run_trino("DROP TABLE IF EXISTS hive.tcdm.web_sales_by_year_gold")

create_gold_sql: str = '''
CREATE TABLE hive.tcdm.web_sales_by_year_gold (
    d_year INTEGER,
    pedidos BIGINT,
    lineas_de_venta BIGINT,
    beneficio_neto DECIMAL(17, 2)
) WITH (
    format = 'PARQUET',
    external_location = 'hdfs://namenode:9000/datalake/gold/tpcds/web_sales_by_year'
)
'''
run_trino(create_gold_sql)
print("tcdm.web_sales_by_year_gold creada.")

run_trino("DESCRIBE hive.tcdm.web_sales_by_year_gold")

gold_df: pd.DataFrame = run_trino("SELECT * FROM hive.tcdm.web_sales_by_year_gold ORDER BY d_year")
gold_df

run_trino("SHOW SCHEMAS FROM hive")

run_trino("SHOW TABLES FROM hive.tcdm")

run_trino("SHOW STATS FOR hive.tcdm.web_sales_silver")

run_trino("ANALYZE hive.tcdm.web_sales_silver")

run_trino("SHOW STATS FOR hive.tcdm.web_sales_silver")

explain_filtered_tras_analyze: pd.DataFrame = run_trino(
    '''EXPLAIN SELECT count(*) FROM hive.tcdm.web_sales_silver
    WHERE sold_year = 2000 AND sold_month = 1'''
)
print("\n".join(explain_filtered_tras_analyze.iloc[:, 0].tolist()))

spark.sql("SHOW TABLES IN tcdm").show()

from pyspark.sql import DataFrame

web_sales_silver_spark: DataFrame = spark.table("tcdm.web_sales_silver")
web_sales_silver_spark.printSchema()

spark.sql('''SELECT sold_year, sold_month, count(*) AS lineas_de_venta
    FROM tcdm.web_sales_silver
    GROUP BY sold_year, sold_month
    ORDER BY sold_year, sold_month''').show(100)

spark.sql("DESCRIBE FORMATTED tcdm.web_sales_silver").show(100, truncate=False)

gold_spark_df: pd.DataFrame = spark.sql(
    "SELECT d_year, pedidos, lineas_de_venta, beneficio_neto FROM tcdm.web_sales_by_year_gold ORDER BY d_year"
).toPandas()
gold_spark_df

comparacion: pd.DataFrame = gold_df.merge(gold_spark_df, on="d_year", suffixes=("_trino", "_spark"))
comparacion["beneficio_neto_trino"] = comparacion["beneficio_neto_trino"].astype(float)
comparacion["beneficio_neto_spark"] = comparacion["beneficio_neto_spark"].astype(float)
comparacion["diferencia_beneficio"] = (
    comparacion["beneficio_neto_trino"] - comparacion["beneficio_neto_spark"]
).abs()

assert (comparacion["pedidos_trino"] == comparacion["pedidos_spark"]).all()
assert (comparacion["lineas_de_venta_trino"] == comparacion["lineas_de_venta_spark"]).all()
assert (comparacion["diferencia_beneficio"] < 0.01).all()
print("Trino y Spark devuelven el mismo resultado de negocio sobre tcdm.web_sales_by_year_gold.")
comparacion

spark.sql("SELECT * FROM tcdm.web_sales_silver WHERE sold_year = 2000 AND sold_month = 1").explain(
    "formatted"
)

spark.catalog.refreshTable("tcdm.web_sales_silver")

run_trino("DROP TABLE IF EXISTS hive.tcdm.web_sales_by_year_managed_demo")

ctas_sql: str = '''
CREATE TABLE hive.tcdm.web_sales_by_year_managed_demo
WITH (format = 'PARQUET') AS
SELECT * FROM hive.tcdm.web_sales_by_year_gold
'''
run_trino(ctas_sql)
print("tcdm.web_sales_by_year_managed_demo creada (tabla administrada).")

run_trino("SHOW CREATE TABLE hive.tcdm.web_sales_by_year_managed_demo")

!hdfs dfs -ls -R -h /warehouse/tcdm.db/web_sales_by_year_managed_demo

!hdfs dfs -mkdir -p /user/luser/s6
!hdfs dfs -rm -r -f /user/luser/s6/drop_demo_external

run_trino("DROP TABLE IF EXISTS hive.tcdm.drop_demo_managed")
run_trino("DROP TABLE IF EXISTS hive.tcdm.drop_demo_external")

run_trino('''
    CREATE TABLE hive.tcdm.drop_demo_managed
    WITH (format = 'PARQUET') AS
    SELECT * FROM hive.tcdm.web_sales_by_year_gold
''')
run_trino('''
    CREATE TABLE hive.tcdm.drop_demo_external
    WITH (
        format = 'PARQUET',
        external_location = 'hdfs://namenode:9000/user/luser/s6/drop_demo_external'
    ) AS
    SELECT * FROM hive.tcdm.web_sales_by_year_gold
''')
print("Tablas efímeras creadas: drop_demo_managed y drop_demo_external.")

!hdfs dfs -ls /warehouse/tcdm.db/drop_demo_managed /user/luser/s6/drop_demo_external

import subprocess

run_trino("DROP TABLE hive.tcdm.drop_demo_managed")
run_trino("DROP TABLE hive.tcdm.drop_demo_external")


def hdfs_dir_exists(path: str) -> bool:
    return subprocess.run(["hdfs", "dfs", "-test", "-d", path]).returncode == 0


managed_dir_exists: bool = hdfs_dir_exists("/warehouse/tcdm.db/drop_demo_managed")
external_dir_exists: bool = hdfs_dir_exists("/user/luser/s6/drop_demo_external")
print(f"¿Sigue el directorio de la tabla administrada? {managed_dir_exists}")
print(f"¿Sigue el directorio de la tabla externa?      {external_dir_exists}")

assert not managed_dir_exists
assert external_dir_exists

!hdfs dfs -ls /user/luser/s6/drop_demo_external

!hdfs dfs -rm -r /user/luser/s6/drop_demo_external

Capa	Responsabilidad	Ejemplo del laboratorio
Almacenamiento	Conserva bytes y rutas	HDFS bajo `/datalake` o `/warehouse`
Formato	Describe cómo leer cada fichero	Parquet
Metastore	Guarda nombres, esquemas, ubicaciones y particiones	Hive Metastore + PostgreSQL
Motor SQL	Planifica y ejecuta consultas	Trino
Motor de transformación	Produce o consume DataFrames	Spark
Formato de tabla	Gestiona metadatos de tabla y cambios	Iceberg en S7

Situación	Orden	Consecuencia
Eliminar sólo la tabla de laboratorio (administrada)	`DROP TABLE hive.tcdm.web_sales_by_year_managed_demo` (desde el cliente Trino de este notebook, o `docker exec trino-hdfs trino --user luser --execute "..."` desde una terminal)	Borra tanto la entrada del metastore como los ficheros Parquet bajo `/warehouse/tcdm.db/web_sales_by_year_managed_demo`: es una tabla administrada, su ciclo de vida de datos depende del catálogo.
Eliminar las definiciones externas de esta sesión	`DROP TABLE hive.tcdm.date_dim`, `DROP TABLE hive.tcdm.web_sales_silver`, `DROP TABLE hive.tcdm.web_sales_by_year_gold`	Borra únicamente las tres entradas del metastore. Los Parquet de `/datalake/raw`, `/datalake/silver` y `/datalake/gold` siguen intactos: nunca interpretes este `DROP TABLE` como "borrar el data lake". S7 necesita las tres definiciones: no las elimines hasta haberla terminado.
Parar temporalmente el warehouse	`docker compose -f entorno/compose-warehouse-hdfs.yml stop`	Conserva PostgreSQL, Hive Metastore y Trino (y su contenido); se reanuda con `start`. No afecta al clúster Hadoop.
Eliminar el warehouse por completo (destructivo)	`docker compose -f entorno/compose-warehouse-hdfs.yml down -v --remove-orphans`	Borra también el volumen de PostgreSQL: se pierde todo el contenido del Hive Metastore, incluidas las definiciones de esta sesión. No borra ni `/datalake` ni `/warehouse` en HDFS: los ficheros Parquet de las tablas externas y los de la tabla administrada de laboratorio (si no se borró antes) seguirían ahí, ahora sin ninguna definición de catálogo que los describa. Tras volver a levantar el warehouse haría falta repetir los `CREATE TABLE` de este notebook.

Sesión 6 — Hive Metastore, tablas Hive y Trino¶

Diapositivas de la sesión¶

Sesión 6: Hive Metastore, tablas Hive y Trino

De la ruta física a la tabla lógica

Capas que se deben distinguir¶

Objetivos de la sesión¶

Cómo se evalúa esta sesión

Antes de empezar¶

Instalar las dependencias de este notebook¶

Dos clientes del mismo metastore

Construir la SparkSession con soporte Hive¶

Abrir la conexión con Trino¶

Registrar sin copiar

Tabla externa frente a tabla administrada¶

Tabla externa sobre el data lake¶

Tabla administrada en el warehouse¶

Registrar los Parquet originales: tcdm.date_dim¶

Registrar no es copiar

Registrar la versión particionada: tcdm.web_sales_silver¶

Por qué hace falta un paso de descubrimiento de particiones¶

Inspeccionar las particiones con $partitions¶

Una partición son dos cosas

Consulta sin filtro frente a consulta filtrada por partición¶

La capa gold y las estadísticas

Registrar la tabla gold: tcdm.web_sales_by_year_gold¶

Consultar con Trino¶

Estadísticas antes y después de ANALYZE¶

Las estadísticas no se calculan solas

Spark como segundo cliente

Consultar la misma tabla con Spark¶

Comparar ubicación y esquema¶

Comparar el resultado de negocio¶

Plan de lectura y caché de metadatos¶

Tabla administrada: tcdm.web_sales_by_year_managed_demo¶

Qué ocurre al borrar cada tipo de tabla¶

Dos ciclos de vida distintos

Buenas prácticas de catálogo

Buenas prácticas de catálogo y data lake¶

Preguntas para interpretar la experiencia¶

Evidencias para la siguiente sesión¶

Qué mostrar en el ordenador durante la reunión¶

Memoria escrita (una o dos páginas)¶

Qué es importante de cara al examen final¶

Detener y limpiar¶

Sesión 6

Siguiente paso¶

Registrar los Parquet originales: `tcdm.date_dim`¶

Registrar la versión particionada: `tcdm.web_sales_silver`¶

Inspeccionar las particiones con `$partitions`¶

Registrar la tabla gold: `tcdm.web_sales_by_year_gold`¶

Estadísticas antes y después de `ANALYZE`¶

Tabla administrada: `tcdm.web_sales_by_year_managed_demo`¶