S2: datos fuente como Parquet sin catálogo
  → S3: los mismos objetos sobre un almacenamiento S3-compatible
  → S4: Spark procesa los ficheros, en local y sobre YARN
  → S5: Spark escribe datos silver y un producto gold
  → S6: el Metastore registra esa organización como tabla Hive

%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

%%writefile requirements.txt
# Dependencias del kernel interactivo de la sesión 5 (Spark en modo local).
# La serie de PySpark aceptada por el curso es >4,<4.2.
pyspark>4,<4.2

%pip install -q -r requirements.txt

import sys

import pyspark

print(f"PySpark {pyspark.__version__} con Python {sys.version.split()[0]}.")

from pyspark import SparkContext
from pyspark.sql import SparkSession

spark: SparkSession = (
    SparkSession.builder.appName("TCDM_S5_Local")
    .master("local[2]")
    .config("spark.driver.memory", "1g")
    .config("spark.sql.shuffle.partitions", "8")
    .getOrCreate()
)
sc: SparkContext = spark.sparkContext

print(f"Spark version: {spark.version} | Master: {sc.master} | App: {sc.appName}")

web_sales/
├── sold_year=1998/
│   ├── sold_month=1/
│   │   └── part-....parquet
│   ├── sold_month=2/
│   │   └── part-....parquet
│   └── ...
├── sold_year=1999/
│   └── ...
└── ...

/datalake/raw/tpcds/web_sales
    → /datalake/silver/tpcds/web_sales/sold_year=.../sold_month=.../part-...

from pyspark.sql.dataframe import DataFrame
from pyspark.sql.functions import col, count, countDistinct
from pyspark.sql.functions import max as spark_max
from pyspark.sql.functions import min as spark_min
from pyspark.sql.functions import month
from pyspark.sql.functions import sum as spark_sum

web_sales_path: str = "/datalake/raw/tpcds/web_sales"
web_sales: DataFrame = spark.read.parquet(web_sales_path)

print(f"Ruta física leída: {web_sales_path}")
web_sales.printSchema()

date_dim_path: str = "/datalake/raw/tpcds/date_dim"
date_dim: DataFrame = spark.read.parquet(date_dim_path)

print(f"Ruta física leída: {date_dim_path}")
date_dim.printSchema()

cardinality_report: DataFrame = web_sales.select(
    countDistinct("ws_order_number").alias("pedidos_distintos"),
    countDistinct("ws_item_sk").alias("articulos_distintos"),
    countDistinct("ws_bill_customer_sk").alias("clientes_distintos"),
)
cardinality_report.show()

web_sales_silver: DataFrame = web_sales.join(
    date_dim.select("d_date_sk", "d_date", "d_year"),
    on=web_sales["ws_sold_date_sk"] == date_dim["d_date_sk"],
    how="inner",
).select(
    "ws_order_number",
    "ws_item_sk",
    "ws_bill_customer_sk",
    "ws_bill_addr_sk",
    "ws_sold_date_sk",
    col("d_date").alias("ws_sold_date"),
    "ws_quantity",
    "ws_list_price",
    "ws_ext_discount_amt",
    "ws_ext_sales_price",
    "ws_net_paid",
    "ws_net_profit",
    col("d_year").alias("sold_year"),
    month(col("d_date")).alias("sold_month"),
)
web_sales_silver.printSchema()

raw_count: int = web_sales.count()
silver_count: int = web_sales_silver.count()
missing_date_key_count: int = web_sales.filter(col("ws_sold_date_sk").isNull()).count()

print(f"Filas en la fuente (web_sales): {raw_count}")
print(f"Filas sin ws_sold_date_sk (no se pueden fechar): {missing_date_key_count}")
print(f"Filas tras la unión con date_dim: {silver_count}")

assert raw_count - silver_count == missing_date_key_count, (
    "El join con date_dim sólo debería perder las líneas sin ws_sold_date_sk, "
    "ninguna otra: "
    f"raw={raw_count} silver={silver_count} sin_fecha={missing_date_key_count}"
)

null_join_keys: int = web_sales_silver.filter(
    col("ws_sold_date_sk").isNull() | col("ws_sold_date").isNull()
).count()

print(f"Líneas con clave de unión o fecha derivada nula: {null_join_keys}")

assert null_join_keys == 0, (
    "No debería haber claves de unión nulas tras un join interno: "
    f"{null_join_keys} filas afectadas"
)

from pyspark.sql import Row

year_month_bounds: Row | None = web_sales_silver.select(
    spark_min("sold_year").alias("min_year"),
    spark_max("sold_year").alias("max_year"),
    spark_min("sold_month").alias("min_month"),
    spark_max("sold_month").alias("max_month"),
).first()

print(year_month_bounds)

assert (
    year_month_bounds["min_month"] >= 1 and year_month_bounds["max_month"] <= 12
), f"sold_month debe estar en el rango 1-12: {year_month_bounds}"

# Las cifras que la sección anterior prometía comparar con las
# cardinalidades descartadas como partición principal.
year_month_combinations: int = web_sales_silver.select("sold_year", "sold_month").distinct().count()
print(f"Combinaciones distintas de sold_year/sold_month: {year_month_combinations}")

silver_path: str = "/datalake/silver/tpcds/web_sales"

(
    web_sales_silver.repartition("sold_year", "sold_month")
    .write.mode("overwrite")
    .partitionBy("sold_year", "sold_month")
    .parquet(silver_path)
)

print(f"Escrito: {silver_path}")

!hdfs dfs -mkdir -p /user/luser/s5

silver_control_path: str = "/user/luser/s5/web_sales_silver_sin_particionar"

web_sales_silver.write.mode("overwrite").parquet(silver_control_path)

print(f"Escrito (control, sin particionar): {silver_control_path}")

web_sales_silver_written: DataFrame = spark.read.parquet(silver_path)

web_sales_by_year: DataFrame = (
    web_sales_silver_written.groupBy("sold_year")
    .agg(
        countDistinct("ws_order_number").alias("pedidos"),
        count("*").alias("lineas_de_venta"),
        spark_sum("ws_net_profit").alias("beneficio_neto"),
    )
    .withColumnRenamed("sold_year", "d_year")
    .orderBy("d_year")
)

web_sales_by_year.printSchema()
web_sales_by_year.show()

gold_path: str = "/datalake/gold/tpcds/web_sales_by_year"

web_sales_by_year.coalesce(1).write.mode("overwrite").parquet(gold_path)

print(f"Escrito: {gold_path}")

web_sales_silver_read_back: DataFrame = spark.read.parquet(silver_path)

web_sales_silver_read_back.printSchema()

year_counts: DataFrame = (
    web_sales_silver_read_back.groupBy("sold_year")
    .count()
    .orderBy(col("count").desc(), col("sold_year"))
)
year_counts.show()

sample_year: int = year_counts.first()["sold_year"]
print(f"Año elegido para el filtro: {sample_year}")

full_scan: DataFrame = web_sales_silver_read_back
filtered: DataFrame = web_sales_silver_read_back.filter(col("sold_year") == sample_year)

full_scan.explain("formatted")

filtered.explain("formatted")

full_scan_rows: int = full_scan.count()
filtered_rows: int = filtered.count()

print(f"Filas de la lectura completa: {full_scan_rows}")
print(f"Filas de la lectura filtrada (sold_year = {sample_year}): {filtered_rows}")

assert filtered_rows < full_scan_rows

import subprocess


def hdfs_file_count(path: str) -> int:
    output: str = subprocess.run(
        ["hdfs", "dfs", "-count", path], check=True, capture_output=True, text=True
    ).stdout
    return int(output.split()[1])


full_scan_files: int = hdfs_file_count(silver_path) - 1  # descarta _SUCCESS
filtered_files: int = hdfs_file_count(f"{silver_path}/sold_year={sample_year}")

print(f"Ficheros bajo la ruta completa: {full_scan_files}")
print(f"Ficheros bajo sold_year={sample_year}: {filtered_files}")

assert filtered_files < full_scan_files, (
    "La partición de un solo año debería tener estrictamente menos ficheros que "
    f"la tabla completa: completa={full_scan_files} filtrada={filtered_files}"
)
print(f"Ficheros descartados por la poda de particiones: {full_scan_files - filtered_files}")

spark.stop()

!hdfs dfs -ls {silver_path}

!hdfs dfs -count -h {silver_path}

!hdfs dfs -ls {silver_path}/sold_year={sample_year}

!hdfs dfs -ls {silver_control_path}

!hdfs dfs -ls {gold_path}

Zona	Función	Aplicación en el curso
Fuente o raw	Datos originales, conservados e inmutables	`/datalake/raw/tpcds`
Refinada o silver	Datos limpiados, enriquecidos y reorganizados	`/datalake/silver/tpcds`
Curada o gold	Productos preparados para una pregunta o consumidor concreto	`/datalake/gold`

Concepto	Significado
Partición de ejecución de Spark	Porción lógica de un DataFrame que procesa una tarea; depende del paralelismo, no de cómo están organizados los ficheros en disco
Bloque HDFS	Unidad de almacenamiento físico de bytes; no tiene por qué coincidir con una fila, una columna ni un fichero completo
Partición Hive (esta sesión)	Directorio `columna=valor` que agrupa físicamente filas por el valor de una o varias columnas
Partición/transformación Iceberg (S7)	Agrupación lógica gestionada por el formato de tabla, que puede usar transformaciones como `month()` sin exigir una columna física propia

Versión	Organización	Filas	Lo que aporta / responde
Fuente (`raw`)	`/datalake/raw/tpcds/web_sales`, un directorio sin particionar	`raw_count` filas (719.384 en SF1)	Qué datos existen físicamente; es la referencia inmutable frente a la que se valida todo lo demás
Refinada sin particionar	`/user/luser/s5/web_sales_silver_sin_particionar`, un directorio, mismas filas que la fuente tras el `join`	`silver_count` filas: `raw_count` menos `missing_date_key_count` (comprobado con `assert`), no `raw_count` exacto	Qué aporta la transformación por sí sola: columnas de negocio limpias, `ws_sold_date` derivada, líneas sin fecha ya descartadas de forma documentada, sin reorganización física todavía
Refinada particionada	`/datalake/silver/tpcds/web_sales/sold_year=.../sold_month=...`, un subdirectorio por combinación año/mes	Las mismas filas que la versión sin particionar, repartidas entre directorios (ver `year_counts` más arriba)	Qué datos puede evitar leer un filtro: compara `full_scan_files` con `filtered_files` — si el segundo es menor, la poda de particiones ha descartado directorios completos
Gold	`/datalake/gold/tpcds/web_sales_by_year`, un único fichero	Una fila por año distinto de `sold_year`	Qué producto consume directamente una pregunta de negocio (pedidos, líneas y beneficio neto por año), sin que quien lo consulte tenga que repetir el `join` con `date_dim`

Sesión 5 — Buenas prácticas de data lake y particionado físico¶

Diapositivas de la sesión¶

Sesión 5: buenas prácticas de data lake y particionado físico

Cómo vamos a recorrer la sesión

Objetivos de la sesión¶

Cómo se evalúa esta sesión

Antes de empezar¶

Instalar PySpark en este notebook¶

Crear una SparkSession local¶

Las convenciones antes del código

Buenas prácticas de data lake¶

Conservar una zona fuente¶

El contrato de silver y gold

Contratos y nombres¶

Qué comprobar antes de dar una salida por buena

Esquema y calidad¶

Un data lake sin catálogo también tiene reglas

Qué significa particionar

Qué significa particionar¶

Elegir bien la columna de partición

Diseño de las particiones¶

Ejercicio de transformación con Spark¶

Leer las tablas fuente¶

Comprobar las cardinalidades antes de elegir la partición¶

Unir con date_dim y construir el esquema silver¶

Validar filas y claves antes de escribir¶

El join validado, no supuesto

Escribir las tres salidas

Escribir la salida silver particionada¶

Escribir una salida de control sin particionar¶

Agregar el producto gold¶

Tres rutas para el mismo dato

La evidencia de que la poda funciona

Releer la tabla silver desde la ruta raíz¶

Comparar una lectura completa con una lectura filtrada¶

Plan físico de la lectura completa¶

Plan físico de la lectura filtrada¶

Ficheros efectivamente leídos¶

Comprobar la estructura de directorios en HDFS¶

Tabla comparativa: fuente, refinada y gold¶

La misma fila, tres organizaciones

Lo que un directorio bien organizado no resuelve

Buenas prácticas que quedan pendientes del catálogo¶

Relación con HDFS y S3¶

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¶

Sesión 5

Siguiente paso¶

El contrato de `silver` y `gold`

Unir con `date_dim` y construir el esquema silver¶