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

%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")

# tpcds.properties
connector.name=tpcds

# hive.properties
connector.name=hive
hive.metastore=thrift
hive.metastore.uri=thrift://hive-metastore:9083
hive.metastore.authentication.type=NONE
hive.metastore.thrift.impersonation.enabled=false
hive.hdfs.authentication.type=NONE
hive.hdfs.impersonation.enabled=false
hive.non-managed-table-creates-enabled=true
hive.non-managed-table-writes-enabled=true
hive.storage-format=PARQUET
hive.compression-codec=ZSTD
fs.hadoop.enabled=true

TPC-DS virtual de Trino          Excel anual del INE
          │                              │
          │                     normalización a CSV UTF-8
          │                              ▼
          │                 raw/ine/municipios/municipios-2026.csv.gz
          │                              │
          └──────── JOIN determinista ───┘
                         │
                         │ CTAS en formato PARQUET
                         ▼
HDFS /datalake/
          ├── raw/ine/municipios/       ← fuente pública versionada
          ├── raw/tpcds/<tabla>/        ← datos de negocio adaptados
          ├── silver/tpcds/<tabla>/     ← datos refinados en Sesión 5
          └── gold/tpcds/<producto>/    ← datos preparados para consumo

HDFS /datalake/raw/tpcds/<tabla>/
          │
          ├── lectura directa con WebHDFS + fsspec + PyArrow
          ├── lectura directa con WebHDFS + fsspec + Polars
          └── lectura SQL con DuckDB

make -C entorno warehouse-up

%%writefile requirements.txt
# Dependencias comunes de los lectores Parquet de la sesión 2.
# WebHDFS de fsspec utiliza requests para sus peticiones HTTP.
duckdb
# DuckDB consulta modified() al registrar WebHDFS como filesystem.
fsspec>=2026.2.0
polars
pyarrow
requests
# Cliente de Trino; se fija la serie mayor, igual que en S6-S8.
trino>=0.330,<1

%pip install -q -r requirements.txt

import getpass
import os
import socket
import sys
from importlib.metadata import version

print(f"Host: {socket.gethostname()}    usuario: {getpass.getuser()}")
print(f"Directorio de trabajo: {os.getcwd()}")
print(f"Intérprete: {sys.executable}")
for package in ("duckdb", "fsspec", "polars", "pyarrow", "requests", "trino"):
    print(f"{package:>10} {version(package)}")

assert socket.gethostname() == "namenode", "El kernel no se ejecuta dentro de namenode"
assert sys.executable.startswith("/opt/tcdm/venv/"), "El kernel no usa /opt/tcdm/venv"

import gzip
from pathlib import Path
from urllib.request import Request, urlopen

# El notebook se ejecuta dentro del contenedor namenode, que no monta el
# directorio entorno/ de la distribución del alumnado: la instantánea se
# descarga del repositorio en vez de asumir una ruta local.
ine_url: str = (
    "https://raw.githubusercontent.com/dsevilla/tcdm-public/26-27/"
    "entorno/data/ine/municipios-2026.csv.gz"
)
compressed_csv: Path = Path("/tmp/ine/municipios-2026.csv.gz")
compressed_csv.parent.mkdir(parents=True, exist_ok=True)

request: Request = Request(ine_url, headers={"User-Agent": "Mozilla/5.0"})
with urlopen(request, timeout=30) as response:
    compressed_csv.write_bytes(response.read())

expected_header: str = (
    "municipio_id;provincia_id;codigo_municipio;" "digito_control;municipio;provincia"
)

# gzip.open descomprime mientras leemos; el CSV no se materializa en disco.
with gzip.open(compressed_csv, "rt", encoding="utf-8", newline="") as source:
    header: str = source.readline().rstrip("\n")
    municipality_count: int = sum(1 for _ in source)

assert header == expected_header
assert municipality_count == 8_132
print(f"Instantánea gzip descargada en {compressed_csv}: " f"{municipality_count:,} municipios")

!hdfs dfs -mkdir -p /datalake/raw/ine/municipios
!hdfs dfs -put -f /tmp/ine/municipios-2026.csv.gz /datalake/raw/ine/municipios/municipios-2026.csv.gz
!hdfs dfs -ls -h /datalake/raw/ine/municipios

from time import sleep

from trino.dbapi import Connection, Cursor, connect

trino_connection: Connection = connect(host="trino-hdfs", port=8080, user="luser")


def trino_sql(sql: str, *, show: bool = True) -> list[list[object]]:
    if show:
        print(f"SQL> {sql}")
    cursor: Cursor = trino_connection.cursor()
    cursor.execute(sql)
    rows: list[list[object]] = cursor.fetchall()
    if show and rows:
        print(*rows[:10], sep="\n")
        if len(rows) > 10:
            print(f"… {len(rows) - 10} filas más")
    return rows


last_error: Exception | None = None
for attempt in range(1, 61):
    try:
        healthcheck: list[list[object]] = trino_sql("SELECT 1", show=False)
        assert len(healthcheck) == 1 and healthcheck[0][0] == 1
        print(f"Trino preparado después de {attempt} intento(s)")
        break
    except Exception as error:
        last_error = error
        sleep(2)
else:
    raise RuntimeError("Trino no respondió durante dos minutos") from last_error

!hdfs dfs -mkdir -p /warehouse/tpcds_bootstrap.db

trino_sql("CREATE SCHEMA IF NOT EXISTS hive.tpcds_bootstrap")
trino_sql("DROP VIEW IF EXISTS hive.tpcds_bootstrap.ine_municipios_ordenados")
trino_sql("DROP TABLE IF EXISTS hive.tpcds_bootstrap.ine_municipios")
trino_sql("""
CREATE TABLE hive.tpcds_bootstrap.ine_municipios (
    municipio_id varchar,
    provincia_id varchar,
    codigo_municipio varchar,
    digito_control varchar,
    municipio varchar,
    provincia varchar
) WITH (
    format = 'CSV',
    csv_separator = ';',
    skip_header_line_count = 1,
    external_location = 'hdfs://namenode:9000/datalake/raw/ine/municipios'
)
""")

# trino_sql() devuelve list[list[object]] porque una fila puede mezclar
# tipos de columna; aquí se sabe que la cuenta es un entero y se
# convierte explícitamente con int() en lugar de dejar un valor object.
municipality_count: int = int(
    trino_sql("SELECT count(*) FROM hive.tpcds_bootstrap.ine_municipios", show=False)[0][0]
)
assert municipality_count == 8_132
print(f"Trino lee {municipality_count:,} municipios")

trino_sql("""
CREATE VIEW hive.tpcds_bootstrap.ine_municipios_ordenados AS
SELECT
    row_number() OVER (ORDER BY municipio_id) AS posicion,
    count(*) OVER () AS total_municipios,
    municipio_id, provincia_id, municipio, provincia
FROM hive.tpcds_bootstrap.ine_municipios
""")

JOIN ine_municipios_ordenados AS m
  ON m.posicion = 1 + mod(
      crc32(to_utf8(concat(
          coalesce(CAST(trim(a.ca_city) AS varchar), ''),
          '|',
          coalesce(CAST(trim(a.ca_state) AS varchar), '')
      ))),
      m.total_municipios
  )

TPCDS_SCALE: str = "sf1"
TPCDS_ROOT: str = "/datalake/raw/tpcds"
LOCALIZED_COLUMNS: dict[str, tuple[str, str, str, str]] = {
    "customer_address": ("ca_city", "ca_county", "ca_state", "ca_country"),
    "store": ("s_city", "s_county", "s_state", "s_country"),
    "warehouse": ("w_city", "w_county", "w_state", "w_country"),
    "call_center": ("cc_city", "cc_county", "cc_state", "cc_country"),
    "web_site": ("web_city", "web_county", "web_state", "web_country"),
}


def localized_select(table_name: str) -> str:
    city, county, state, country = LOCALIZED_COLUMNS[table_name]
    described: list[list[object]] = trino_sql(
        f"DESCRIBE tpcds.{TPCDS_SCALE}.{table_name}", show=False
    )
    column_types: dict[object, object] = {row[0]: row[1] for row in described}
    replacements: dict[str, str] = {
        city: f"CAST(m.municipio AS {column_types[city]}) AS {city}",
        county: f"CAST(m.provincia AS {column_types[county]}) AS {county}",
        state: f"CAST(m.provincia_id AS {column_types[state]}) AS {state}",
        country: f"CAST('España' AS {column_types[country]}) AS {country}",
    }
    projection: str = ",\n    ".join(
        replacements.get(column, f"a.{column}") for column, *_ in described
    )
    return f"""
SELECT
    {projection}
FROM tpcds.{TPCDS_SCALE}.{table_name} AS a
JOIN hive.tpcds_bootstrap.ine_municipios_ordenados AS m
  ON m.posicion = 1 + mod(
      crc32(to_utf8(concat(
          coalesce(CAST(trim(a.{city}) AS varchar), ''),
          '|',
          coalesce(CAST(trim(a.{state}) AS varchar), '')
      ))),
      m.total_municipios
  )
""".strip()


print(localized_select("customer_address"))

from subprocess import DEVNULL, PIPE, CompletedProcess, run


def hdfs(*arguments: str, check: bool = True, capture: bool = False) -> CompletedProcess[str]:
    return run(
        ["hdfs", "dfs", *arguments],
        check=check,
        text=True,
        stdout=PIPE if capture else None,
        stderr=None if check else DEVNULL,
    )

# trino_sql() devuelve list[list[object]]; aquí se sabe que la primera
# columna es un nombre de tabla y se convierte explícitamente con str()
# en lugar de dejar un valor de tipo object.
table_names: list[str] = sorted(
    str(row[0])
    for row in trino_sql(f"SHOW TABLES FROM tpcds.{TPCDS_SCALE}", show=False)
    if row[0] != "dbgen_version"
)
assert len(table_names) == 24, table_names

materialized: list[str] = []
already_available: list[str] = []
success_file: Path = Path("/tmp/tpcds-success-marker")

for table_name in table_names:
    table_path: str = f"{TPCDS_ROOT}/{table_name}"
    marker_path: str = f"{table_path}/_SUCCESS"
    expected_marker: str = "INE-2026" if table_name in LOCALIZED_COLUMNS else ""
    marker_exists: bool = hdfs("-test", "-e", marker_path, check=False).returncode == 0
    if marker_exists:
        marker: str = hdfs("-cat", marker_path, capture=True).stdout
        if marker == expected_marker:
            # Una ejecución interrumpida pudo dejar metadatos temporales
            # aunque los Parquet y su marcador ya estuvieran completos.
            trino_sql(f"DROP TABLE IF EXISTS hive.tpcds_bootstrap.{table_name}", show=False)
            already_available.append(table_name)
            print(f"[=] {table_name}: ya está materializada")
            continue

    print(f"[*] {table_name}: generando Parquet")
    trino_sql(f"DROP TABLE IF EXISTS hive.tpcds_bootstrap.{table_name}", show=False)
    hdfs("-rm", "-r", "-f", table_path, check=False)
    select_query: str = (
        localized_select(table_name)
        if table_name in LOCALIZED_COLUMNS
        else f"SELECT * FROM tpcds.{TPCDS_SCALE}.{table_name}"
    )
    trino_sql(
        f"""
CREATE TABLE hive.tpcds_bootstrap.{table_name}
WITH (
    format = 'PARQUET',
    external_location = 'hdfs://namenode:9000{table_path}'
) AS
{select_query}
""",
        show=False,
    )
    success_file.write_text(expected_marker, encoding="utf-8")
    hdfs("-put", "-f", str(success_file), marker_path)
    trino_sql(f"DROP TABLE hive.tpcds_bootstrap.{table_name}", show=False)
    materialized.append(table_name)

print(f"Materializadas ahora: {len(materialized)}; reutilizadas: {len(already_available)}")

for table_name in table_names:
    marker_path: str = f"{TPCDS_ROOT}/{table_name}/_SUCCESS"
    assert hdfs("-test", "-e", marker_path, check=False).returncode == 0
    if table_name in LOCALIZED_COLUMNS:
        assert hdfs("-cat", marker_path, capture=True).stdout == "INE-2026"

trino_sql("DROP VIEW hive.tpcds_bootstrap.ine_municipios_ordenados", show=False)
trino_sql("DROP TABLE hive.tpcds_bootstrap.ine_municipios", show=False)
trino_sql("DROP SCHEMA IF EXISTS hive.tpcds_bootstrap", show=False)

print("Dataset preparado: 24 tablas TPC-DS y fuente municipal del INE en HDFS")
!hdfs dfs -du -h -s /datalake/raw/tpcds

!hdfs dfs -ls -h /datalake/raw/ine/municipios
!hdfs dfs -text /datalake/raw/ine/municipios/municipios-2026.csv.gz | sed -n '1,6p'

from pathlib import Path
from urllib.request import Request, urlopen

ine_url = "https://www.ine.es/daco/daco42/codmun/26codmun.xlsx"
ine_excel = Path("/tmp/26codmun.xlsx")

request = Request(ine_url, headers={"User-Agent": "Mozilla/5.0"})
with urlopen(request, timeout=30) as response:
    ine_excel.write_bytes(response.read())

print(f"Descargados {ine_excel.stat().st_size:,} bytes")

import polars as pl

sheets = pl.read_excel(ine_excel, sheet_id=0, has_header=False)
first_sheet = next(iter(sheets.values()))

print(f"{len(sheets)} hojas provinciales")
first_sheet.head()

expected_header = ["CPRO", "CMUN", "DC", "NOMBRE"]
municipalities: list[tuple[str, ...]] = []

for frame in sheets.values():
    header = [str(value) for value in frame.row(2)[:4]]
    assert header == expected_header
    province = str(frame.row(1)[0])
    for row in frame[3:].iter_rows():
        province_id, municipality_code, check_digit, municipality = (
            "" if value is None else str(value) for value in row[:4]
        )
        if not all((province_id, municipality_code, check_digit, municipality)):
            continue
        municipality_id = f"{province_id}{municipality_code}"
        municipalities.append(
            (municipality_id, province_id, municipality_code, check_digit, municipality, province)
        )

municipalities.sort(key=lambda row: row[0])
municipalities[:5]

municipality_ids = [row[0] for row in municipalities]
province_ids = {row[1] for row in municipalities}

assert len(municipalities) == 8_132
assert len(set(municipality_ids)) == len(municipality_ids)
assert province_ids == {f"{code:02d}" for code in range(1, 53)}
print("8.132 municipios únicos y códigos provinciales 01–52")

import csv
import gzip
import shutil
from pathlib import Path

regenerated_csv = Path("/tmp/municipios-2026.csv")
with regenerated_csv.open("w", encoding="utf-8", newline="") as output:
    writer = csv.writer(output, delimiter=";", lineterminator="\n")
    writer.writerow(
        (
            "municipio_id",
            "provincia_id",
            "codigo_municipio",
            "digito_control",
            "municipio",
            "provincia",
        )
    )
    writer.writerows(municipalities)

regenerated_gzip = Path("/tmp/municipios-2026.csv.gz")
with regenerated_csv.open("rb") as source, regenerated_gzip.open("wb") as raw_output:
    with gzip.GzipFile(
        filename="",
        mode="wb",
        fileobj=raw_output,
        compresslevel=9,
        mtime=0,
    ) as destination:
        shutil.copyfileobj(source, destination)

!hdfs dfs -ls -h /datalake/raw

!hdfs dfs -ls -h /datalake/raw/tpcds/date_dim

!hdfs dfs -du -h /datalake/raw/tpcds/date_dim

!hdfs dfs -cat $(hdfs dfs -ls /datalake/raw/tpcds/date_dim \
  | awk '$NF ~ /date_dim\// && $NF !~ /_SUCCESS$/ { print $NF; exit }') 2>/dev/null \
  | od -An -c -N 4

!hdfs dfs -du -h /datalake/raw/tpcds

!hdfs dfs -du -h -s /datalake/raw/tpcds

!hdfs fsck /datalake/raw/tpcds/date_dim -files -blocks -locations

import fsspec
from fsspec.spec import AbstractFileSystem

webhdfs: AbstractFileSystem = fsspec.filesystem(
    "webhdfs",
    host="namenode",
    port=9870,
    user="luser",
    use_https=False,
)

table_path: str = "/datalake/raw/tpcds/date_dim"
files: list[str] = sorted(
    path for path in webhdfs.glob(f"{table_path}/*") if not path.endswith("/_SUCCESS")
)

print(f"Ficheros de datos: {len(files)}")
print(files[:3])

from typing import Any

info: dict[str, Any] = webhdfs.info(files[0])
print(info["name"])
print(info["size"], "bytes")

import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.parquet as pq

arrow_fs: pa.fs.FileSystem = pa.fs.PyFileSystem(pa.fs.FSSpecHandler(webhdfs))
arrow_table: pa.Table = pq.read_table(files, filesystem=arrow_fs)

key_values: pa.ChunkedArray = arrow_table["d_date_sk"]
print(arrow_table.schema)
summary = (
    arrow_table.num_rows,
    pc.count(key_values).as_py(),
    pc.min(key_values).as_py(),
    pc.max(key_values).as_py(),
    pc.sum(key_values).as_py(),
)
print(summary)
print("TCDM_SUMMARY\tpyarrow\tdate_dim\t" + "\t".join(map(str, summary)))

import polars as pl

fragments: list[pl.DataFrame] = [pl.read_parquet(webhdfs.open(path, "rb")) for path in files]
polars_table: pl.DataFrame = pl.concat(fragments, how="vertical")

key_values: pl.Series = polars_table.get_column("d_date_sk")
print(polars_table.schema)
summary = (
    polars_table.height,
    key_values.len() - key_values.null_count(),
    key_values.min(),
    key_values.max(),
    key_values.sum(),
)
print(summary)
print("TCDM_SUMMARY\tpolars\tdate_dim\t" + "\t".join(map(str, summary)))

import duckdb

connection: duckdb.DuckDBPyConnection = duckdb.connect()
duckdb.register_filesystem(webhdfs, connection=connection)
urls: list[str] = [f"webhdfs://{path}" for path in files]

description: list[tuple[object, ...]] = connection.execute(
    "DESCRIBE SELECT * FROM read_parquet(?)", [urls]
).fetchall()
print(description)

summary: tuple[object, ...] | None = connection.execute(
    """
    SELECT count(*), count(d_date_sk), min(d_date_sk), max(d_date_sk), sum(d_date_sk)
    FROM read_parquet(?)
    """,
    [urls],
).fetchone()
print(summary)
print("TCDM_SUMMARY\tduckdb\tdate_dim\t" + "\t".join(map(str, summary)))

import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.parquet as pq

arrow_fs: pa.fs.FileSystem = pa.fs.PyFileSystem(pa.fs.FSSpecHandler(webhdfs))

arrow_table: pa.Table = pq.read_table(
    files,
    filesystem=arrow_fs,
    columns=["d_date", "d_year", "d_holiday"],
)

print(arrow_table)
print(arrow_table.num_rows)

recent_dates: pa.Table = arrow_table.filter(
    pc.greater_equal(
        arrow_table["d_year"],
        1998,
    )
)
print(recent_dates)

print(arrow_table.schema)

import polars as pl

columns: list[str] = ["d_date", "d_year", "d_holiday"]
fragments: list[pl.DataFrame] = [
    pl.read_parquet(webhdfs.open(path, "rb"), columns=columns) for path in files
]
selected_dates: pl.DataFrame = pl.concat(fragments, how="vertical")

print(selected_dates.schema)
print(selected_dates.height)

import polars as pl

dates: pl.DataFrame = pl.from_arrow(arrow_table)
print(dates)
print(dates.schema)

recent_dates: pl.DataFrame = dates.filter(pl.col("d_year") >= 1998).select(
    ["d_date", "d_year", "d_holiday"]
)
print(recent_dates)

with webhdfs.open(files[0], "rb") as source:
    first_fragment: pl.DataFrame = pl.read_parquet(
        source,
        columns=["d_date", "d_year"],
    )

print(first_fragment.shape)

fragments: list[pl.DataFrame] = []
for path in files:
    with webhdfs.open(path, "rb") as source:
        fragments.append(pl.read_parquet(source, columns=["d_date", "d_year"]))

all_dates: pl.DataFrame = pl.concat(fragments)
print(all_dates.shape)

webhdfs_files: list[str] = sorted(
    path
    for path in webhdfs.glob("/datalake/raw/tpcds/date_dim/*")
    if not path.endswith("/_SUCCESS")
)
urls: list[str] = [f"webhdfs://{path}" for path in webhdfs_files]

result: list[tuple[object, ...]] = connection.execute(
    "SELECT count(*) AS rows FROM read_parquet(?)",
    [urls],
).fetchall()
print(result)

result: list[tuple[object, ...]] = connection.execute(
    """
    SELECT d_year, d_holiday, count(*) AS days
    FROM read_parquet(?)
    WHERE d_year BETWEEN 1998 AND 2000
    GROUP BY d_year, d_holiday
    ORDER BY d_year, d_holiday
    """,
    [urls],
).fetchall()

print(result)

arrow_result: pa.Table = connection.execute(
    "SELECT d_year, count(*) AS days FROM read_parquet(?) GROUP BY d_year",
    [urls],
).fetch_arrow_table()
print(pl.from_arrow(arrow_result))

import polars as pl
import pyarrow as pa

duckdb_web_sales_files: list[str] = sorted(
    path
    for path in webhdfs.glob("/datalake/raw/tpcds/web_sales/*")
    if not path.endswith("/_SUCCESS")
)
duckdb_web_sales_urls: list[str] = [f"webhdfs://{path}" for path in duckdb_web_sales_files]

print("Orden: listar los ficheros Parquet de web_sales")
print(*duckdb_web_sales_files, sep="\n")

print("Orden: describir la tabla virtual creada por read_parquet")
schema_rows: list[tuple[object, ...]] = connection.execute(
    "DESCRIBE SELECT * FROM read_parquet(?)",
    [duckdb_web_sales_urls],
).fetchall()
print(*schema_rows, sep="\n")

print("Orden: leer cinco filas de web_sales")
sample: pa.Table = connection.execute(
    "SELECT * FROM read_parquet(?) LIMIT 5",
    [duckdb_web_sales_urls],
).fetch_arrow_table()
print(sample)
print("La misma muestra convertida a un DataFrame Polars:")
sample_polars: pl.DataFrame = pl.from_arrow(sample)
print(sample_polars)

duckdb_customer_files: list[str] = sorted(
    path
    for path in webhdfs.glob("/datalake/raw/tpcds/customer/*")
    if not path.endswith("/_SUCCESS")
)
duckdb_customer_urls: list[str] = [f"webhdfs://{path}" for path in duckdb_customer_files]

print("Orden: comprobar la FK web_sales.ws_bill_customer_sk")
orphan_customer_count: int = int(
    connection.execute(
        """
    SELECT count(*)
    FROM read_parquet(?) AS sales
    LEFT JOIN read_parquet(?) AS customers
      ON sales.ws_bill_customer_sk = customers.c_customer_sk
    WHERE sales.ws_bill_customer_sk IS NOT NULL
      AND customers.c_customer_sk IS NULL
    """,
        [duckdb_web_sales_urls, duckdb_customer_urls],
    ).fetchone()[0]
)
print(f"Filas de web_sales sin cliente correspondiente: {orphan_customer_count}")

connection.execute("DROP TABLE IF EXISTS demo_web_sales")
connection.execute("DROP TABLE IF EXISTS demo_customer")

print("Orden: crear la tabla temporal de clientes con PK")
connection.execute("""
    CREATE TEMPORARY TABLE demo_customer (
        c_customer_sk BIGINT PRIMARY KEY
    )
    """)
connection.execute(
    """
    INSERT INTO demo_customer
    SELECT c_customer_sk
    FROM read_parquet(?)
    """,
    [duckdb_customer_urls],
)

print("Orden: crear la tabla temporal de ventas con FK")
connection.execute("""
    CREATE TEMPORARY TABLE demo_web_sales (
        ws_item_sk BIGINT,
        ws_order_number BIGINT,
        ws_bill_customer_sk BIGINT,
        FOREIGN KEY (ws_bill_customer_sk)
            REFERENCES demo_customer (c_customer_sk)
    )
    """)
connection.execute(
    """
    INSERT INTO demo_web_sales
    SELECT ws_item_sk, ws_order_number, ws_bill_customer_sk
    FROM read_parquet(?)
    WHERE ws_bill_customer_sk IS NOT NULL
    LIMIT 1000
    """,
    [duckdb_web_sales_urls],
)

print("Orden: consultar las restricciones declaradas")
constraint_rows: list[tuple[object, ...]] = connection.execute("""
    SELECT table_name, constraint_type, constraint_name
    FROM information_schema.table_constraints
    WHERE table_name IN ('demo_customer', 'demo_web_sales')
    ORDER BY table_name, constraint_type
    """).fetchall()
print(*constraint_rows, sep="\n")

try:
    connection.execute("INSERT INTO demo_web_sales VALUES (1, 1, -1)")
except duckdb.ConstraintException as error:
    print(f"DuckDB rechaza la fila: {error}")
else:
    raise AssertionError("La clave ajena debería haber rechazado la inserción")

def table_urls(table: str) -> list[str]:
    return [
        f"webhdfs://{path}"
        for path in sorted(webhdfs.glob(f"/datalake/raw/tpcds/{table}/*"))
        if not path.endswith("/_SUCCESS")
    ]


def table_sample(table: str, columns: list[str], rows: int = 5) -> pl.DataFrame:
    selection: str = ", ".join(columns)
    sample_rows: pa.Table = connection.execute(
        f"SELECT {selection} FROM read_parquet(?) LIMIT {rows}",
        [table_urls(table)],
    ).fetch_arrow_table()
    return pl.from_arrow(sample_rows).with_columns(pl.col(pl.String).str.strip_chars())

print(
    table_sample(
        "customer",
        ["c_customer_sk", "c_first_name", "c_last_name", "c_birth_year", "c_current_addr_sk"],
    )
)
print(
    table_sample(
        "customer_address",
        ["ca_address_sk", "ca_street_name", "ca_city", "ca_county", "ca_state"],
    )
)

print(
    table_sample(
        "item", ["i_item_sk", "i_product_name", "i_category", "i_brand", "i_current_price"]
    )
)
print(table_sample("date_dim", ["d_date_sk", "d_date", "d_year", "d_moy", "d_day_name"]))
print(table_sample("time_dim", ["t_time_sk", "t_hour", "t_minute", "t_am_pm", "t_shift"]))
print(table_sample("store", ["s_store_sk", "s_store_name", "s_city", "s_county"]))

print(
    table_sample(
        "web_sales",
        [
            "ws_order_number",
            "ws_item_sk",
            "ws_bill_customer_sk",
            "ws_bill_addr_sk",
            "ws_sold_date_sk",
            "ws_quantity",
            "ws_net_profit",
        ],
    )
)
print(
    table_sample(
        "store_sales",
        [
            "ss_ticket_number",
            "ss_item_sk",
            "ss_customer_sk",
            "ss_store_sk",
            "ss_quantity",
            "ss_net_profit",
        ],
    )
)

readable_sales: pa.Table = connection.execute(
    """
    SELECT
        s.ws_order_number,
        trim(i.i_product_name) AS producto,
        trim(c.c_first_name) || ' ' || trim(c.c_last_name) AS cliente,
        trim(a.ca_city) AS municipio,
        d.d_date AS fecha,
        s.ws_quantity,
        s.ws_net_profit
    FROM read_parquet(?) AS s
    JOIN read_parquet(?) AS i ON s.ws_item_sk = i.i_item_sk
    JOIN read_parquet(?) AS c ON s.ws_bill_customer_sk = c.c_customer_sk
    JOIN read_parquet(?) AS a ON s.ws_bill_addr_sk = a.ca_address_sk
    JOIN read_parquet(?) AS d ON s.ws_sold_date_sk = d.d_date_sk
    LIMIT 5
    """,
    [
        table_urls("web_sales"),
        table_urls("item"),
        table_urls("customer"),
        table_urls("customer_address"),
        table_urls("date_dim"),
    ],
).fetch_arrow_table()
print(pl.from_arrow(readable_sales))

web_sales_root: str = "/datalake/raw/tpcds/web_sales"
customer_root: str = "/datalake/raw/tpcds/customer"
address_root: str = "/datalake/raw/tpcds/customer_address"
time_root: str = "/datalake/raw/tpcds/time_dim"

web_sales_files: list[str] = sorted(
    path for path in webhdfs.glob(f"{web_sales_root}/*") if not path.endswith("/_SUCCESS")
)
customer_files: list[str] = sorted(
    path for path in webhdfs.glob(f"{customer_root}/*") if not path.endswith("/_SUCCESS")
)
address_files: list[str] = sorted(
    path for path in webhdfs.glob(f"{address_root}/*") if not path.endswith("/_SUCCESS")
)
time_files: list[str] = sorted(
    path for path in webhdfs.glob(f"{time_root}/*") if not path.endswith("/_SUCCESS")
)

web_sales_urls: list[str] = [f"webhdfs://{path}" for path in web_sales_files]
customer_urls: list[str] = [f"webhdfs://{path}" for path in customer_files]
time_urls: list[str] = [f"webhdfs://{path}" for path in time_files]

print(len(web_sales_files), "ficheros de web_sales")
print(len(customer_files), "ficheros de customer")
print(len(address_files), "ficheros de customer_address")
print(len(time_files), "ficheros de time_dim")

sales_for_customer: pa.Table = pq.read_table(
    web_sales_files,
    filesystem=arrow_fs,
    columns=["ws_bill_customer_sk", "ws_order_number"],
)
sales_for_customer = sales_for_customer.filter(
    pc.is_valid(sales_for_customer["ws_bill_customer_sk"])
)

top_customer_keys: pa.Table = (
    sales_for_customer.group_by("ws_bill_customer_sk")
    .aggregate([("ws_order_number", "count_distinct")])
    .rename_columns(["ws_bill_customer_sk", "orders"])
    .sort_by([("orders", "descending")])
    .slice(0, 10)
)

print(top_customer_keys)

sales_parts: list[pl.DataFrame] = []
for path in web_sales_files:
    with webhdfs.open(path, "rb") as source:
        sales_parts.append(
            pl.read_parquet(
                source,
                columns=[
                    "ws_bill_addr_sk",
                    "ws_order_number",
                    "ws_ext_sales_price",
                ],
            )
        )
sales_for_province: pl.DataFrame = pl.concat(sales_parts, how="vertical")

address_parts: list[pl.DataFrame] = []
for path in address_files:
    with webhdfs.open(path, "rb") as source:
        address_parts.append(
            pl.read_parquet(
                source,
                columns=["ca_address_sk", "ca_city", "ca_county", "ca_state"],
            )
        )
addresses: pl.DataFrame = pl.concat(address_parts, how="vertical")

province_summary: pl.DataFrame = (
    sales_for_province.join(
        addresses,
        left_on="ws_bill_addr_sk",
        right_on="ca_address_sk",
        how="left",
    )
    .filter(pl.col("ca_county").is_not_null())
    .group_by(["ca_state", "ca_county"])
    .agg(
        pl.col("ws_order_number").n_unique().alias("orders"),
        pl.col("ws_ext_sales_price").sum().alias("sales"),
    )
    .sort(["orders", "ca_state"], descending=[True, False])
    .head(10)
)

print(province_summary)

import gzip

ine_path: str = "/datalake/raw/ine/municipios/municipios-2026.csv.gz"
with webhdfs.open(ine_path, "rb") as source, gzip.GzipFile(fileobj=source) as decompressed:
    ine_municipalities: pl.DataFrame = pl.read_csv(
        decompressed,
        separator=";",
        schema_overrides={
            "municipio_id": pl.String,
            "provincia_id": pl.String,
            "codigo_municipio": pl.String,
            "digito_control": pl.String,
        },
    )

address_places: pl.DataFrame = addresses.select(
    pl.col("ca_city").str.strip_chars().alias("municipio"),
    pl.col("ca_county").str.strip_chars().alias("provincia"),
    pl.col("ca_state").str.strip_chars().alias("provincia_id"),
).unique()

unknown_places: pl.DataFrame = address_places.join(
    ine_municipalities.select("municipio", "provincia", "provincia_id"),
    on=["municipio", "provincia", "provincia_id"],
    how="anti",
)

print(f"Municipios del catálogo INE: {ine_municipalities.height}")
print(f"Lugares distintos usados en customer_address: {address_places.height}")
print(f"Lugares sin correspondencia en el INE: {unknown_places.height}")
assert ine_municipalities.height == 8_132
assert unknown_places.is_empty()

from decimal import Decimal

hourly_sales: list[tuple[int, int, Decimal | None]] = connection.execute(
    """
    SELECT
        t.t_hour,
        count(DISTINCT s.ws_order_number) AS orders,
        sum(s.ws_ext_sales_price) AS sales
    FROM read_parquet(?) AS s
    JOIN read_parquet(?) AS t
        ON s.ws_sold_time_sk = t.t_time_sk
    GROUP BY t.t_hour
    ORDER BY t.t_hour
    """,
    [web_sales_urls, time_urls],
).fetchall()

for hour, orders, sales in hourly_sales:
    print(f"hora={hour:02d} pedidos={orders} ventas={sales}")

top_customers: list[tuple[str, int, Decimal | None]] = connection.execute(
    """
    SELECT
        c.c_customer_id,
        count(DISTINCT s.ws_order_number) AS orders,
        sum(s.ws_ext_sales_price) AS sales
    FROM read_parquet(?) AS s
    JOIN read_parquet(?) AS c
        ON s.ws_bill_customer_sk = c.c_customer_sk
    GROUP BY c.c_customer_id
    ORDER BY orders DESC, c.c_customer_id
    LIMIT 10
    """,
    [web_sales_urls, customer_urls],
).fetchall()

for customer_id, orders, sales in top_customers:
    print(f"cliente={customer_id} pedidos={orders} ventas={sales}")

def data_files(table: str) -> list[str]:
    root: str = f"/datalake/raw/tpcds/{table}"
    return sorted(path for path in webhdfs.glob(f"{root}/*") if not path.endswith("/_SUCCESS"))


sales_urls: list[str] = [f"webhdfs://{path}" for path in data_files("store_sales")]
date_urls: list[str] = [f"webhdfs://{path}" for path in data_files("date_dim")]
item_urls: list[str] = [f"webhdfs://{path}" for path in data_files("item")]

query: str = """
SELECT
    d.d_year,
    i.i_category,
    sum(s.ss_ext_sales_price) AS sales
FROM read_parquet(?) AS s
JOIN read_parquet(?) AS d
    ON s.ss_sold_date_sk = d.d_date_sk
JOIN read_parquet(?) AS i
    ON s.ss_item_sk = i.i_item_sk
WHERE d.d_year BETWEEN 1998 AND 2000
GROUP BY d.d_year, i.i_category
ORDER BY d.d_year, sales DESC
"""

sales_by_category: list[tuple[object, ...]] = connection.execute(
    query,
    [sales_urls, date_urls, item_urls],
).fetchall()
print(sales_by_category[:10])

store_sales_urls: list[str] = table_urls("store_sales")
catalog_sales_urls: list[str] = table_urls("catalog_sales")
address_urls: list[str] = table_urls("customer_address")

print(len(store_sales_urls), "ficheros de store_sales")
print(len(catalog_sales_urls), "ficheros de catalog_sales")
print(len(address_urls), "ficheros de customer_address")

print(
    table_sample(
        "catalog_sales",
        [
            "cs_order_number",
            "cs_item_sk",
            "cs_call_center_sk",
            "cs_catalog_page_sk",
            "cs_quantity",
            "cs_ext_sales_price",
        ],
    )
)

sales_by_channel: pa.Table = connection.execute(
    """
    SELECT
        'tienda' AS canal,
        count(*) AS lineas,
        sum(ss_quantity) AS unidades,
        sum(ss_ext_sales_price) AS ventas
    FROM read_parquet(?)
    UNION ALL
    SELECT 'catálogo', count(*), sum(cs_quantity), sum(cs_ext_sales_price)
    FROM read_parquet(?)
    UNION ALL
    SELECT 'web', count(*), sum(ws_quantity), sum(ws_ext_sales_price)
    FROM read_parquet(?)
    ORDER BY ventas DESC
    """,
    [store_sales_urls, catalog_sales_urls, web_sales_urls],
).fetch_arrow_table()
print(pl.from_arrow(sales_by_channel))

item_levels: pa.Table = connection.execute(
    """
    SELECT
        count(*) AS filas,
        count(DISTINCT i_item_id) AS identificadores,
        count(DISTINCT i_category) AS categorias,
        count(DISTINCT i_class) AS clases,
        count(DISTINCT i_brand) AS marcas
    FROM read_parquet(?)
    """,
    [item_urls],
).fetch_arrow_table()
print(pl.from_arrow(item_levels))

sales_by_item_category: list[tuple[str | None, int, Decimal]] = connection.execute(
    """
    SELECT
        trim(i.i_category) AS categoria,
        sum(s.ss_quantity) AS unidades,
        sum(s.ss_ext_sales_price) AS ventas
    FROM read_parquet(?) AS s
    JOIN read_parquet(?) AS i
        ON s.ss_item_sk = i.i_item_sk
    GROUP BY categoria
    ORDER BY ventas DESC
    """,
    [store_sales_urls, item_urls],
).fetchall()

for category, units, sales in sales_by_item_category:
    print(f"categoria={category} unidades={units} ventas={sales}")

top_items: pa.Table = connection.execute(
    """
    SELECT
        i.i_item_sk,
        trim(i.i_product_name) AS producto,
        trim(i.i_category) AS categoria,
        sum(s.ss_quantity) AS unidades,
        sum(s.ss_ext_sales_price) AS ventas
    FROM read_parquet(?) AS s
    JOIN read_parquet(?) AS i
        ON s.ss_item_sk = i.i_item_sk
    GROUP BY i.i_item_sk, i.i_product_name, i.i_category
    ORDER BY unidades DESC, i.i_item_sk
    LIMIT 10
    """,
    [store_sales_urls, item_urls],
).fetch_arrow_table()
print(pl.from_arrow(top_items))

from datetime import date

sales_period: list[tuple[int, int, date, date]] = connection.execute(
    """
    SELECT
        d.d_year,
        count(*) AS lineas,
        min(d.d_date) AS primera,
        max(d.d_date) AS ultima
    FROM read_parquet(?) AS s
    JOIN read_parquet(?) AS d
        ON s.ss_sold_date_sk = d.d_date_sk
    GROUP BY d.d_year
    ORDER BY d.d_year
    """,
    [store_sales_urls, date_urls],
).fetchall()

for year, sale_lines, first_day, last_day in sales_period:
    print(f"año={year} lineas={sale_lines} desde={first_day} hasta={last_day}")

monthly_sales: list[tuple[int, int, int, Decimal]] = connection.execute(
    """
    SELECT
        d.d_moy AS mes,
        count(DISTINCT s.ss_ticket_number) AS tickets,
        sum(s.ss_quantity) AS unidades,
        sum(s.ss_ext_sales_price) AS ventas
    FROM read_parquet(?) AS s
    JOIN read_parquet(?) AS d
        ON s.ss_sold_date_sk = d.d_date_sk
    WHERE d.d_year BETWEEN 1998 AND 2002
    GROUP BY d.d_moy
    ORDER BY d.d_moy
    """,
    [store_sales_urls, date_urls],
).fetchall()

for month, tickets, units, sales in monthly_sales:
    print(f"mes={month:02d} tickets={tickets} unidades={units} ventas={sales}")

top_item_by_month: list[tuple[int, str, str | None, int]] = connection.execute(
    """
    WITH unidades_por_mes AS (
        SELECT
            d.d_moy AS mes,
            i.i_item_sk,
            trim(i.i_product_name) AS producto,
            trim(i.i_category) AS categoria,
            sum(s.ss_quantity) AS unidades
        FROM read_parquet(?) AS s
        JOIN read_parquet(?) AS d
            ON s.ss_sold_date_sk = d.d_date_sk
        JOIN read_parquet(?) AS i
            ON s.ss_item_sk = i.i_item_sk
        WHERE d.d_year BETWEEN 1998 AND 2002
        GROUP BY d.d_moy, i.i_item_sk, i.i_product_name, i.i_category
    ),
    clasificados AS (
        SELECT
            *,
            row_number() OVER (
                PARTITION BY mes
                ORDER BY unidades DESC, i_item_sk
            ) AS posicion
        FROM unidades_por_mes
    )
    SELECT mes, producto, categoria, unidades
    FROM clasificados
    WHERE posicion = 1
    ORDER BY mes
    """,
    [store_sales_urls, date_urls, item_urls],
).fetchall()

for month, product, category, units in top_item_by_month:
    print(f"mes={month:02d} producto={product} categoria={category} unidades={units}")

address_levels: pa.Table = connection.execute(
    """
    SELECT
        count(*) AS domicilios,
        count(DISTINCT ca_state) AS provincias,
        count(DISTINCT (ca_state, ca_city)) AS municipios
    FROM read_parquet(?)
    """,
    [address_urls],
).fetch_arrow_table()
print(pl.from_arrow(address_levels))

lines_with_address: pa.Table = connection.execute(
    """
    SELECT
        count(*) AS lineas,
        count(ss_addr_sk) AS con_direccion
    FROM read_parquet(?)
    """,
    [store_sales_urls],
).fetch_arrow_table()
print(pl.from_arrow(lines_with_address))

top_item_by_province: pa.Table = connection.execute(
    """
    WITH unidades_por_provincia AS (
        SELECT
            trim(a.ca_county) AS provincia,
            i.i_item_sk,
            trim(i.i_product_name) AS producto,
            trim(i.i_category) AS categoria,
            sum(s.ss_quantity) AS unidades
        FROM read_parquet(?) AS s
        JOIN read_parquet(?) AS a
            ON s.ss_addr_sk = a.ca_address_sk
        JOIN read_parquet(?) AS i
            ON s.ss_item_sk = i.i_item_sk
        GROUP BY a.ca_county, i.i_item_sk, i.i_product_name, i.i_category
    ),
    clasificados AS (
        SELECT
            *,
            row_number() OVER (
                PARTITION BY provincia
                ORDER BY unidades DESC, i_item_sk
            ) AS posicion
        FROM unidades_por_provincia
    )
    SELECT provincia, producto, categoria, unidades
    FROM clasificados
    WHERE posicion = 1
    ORDER BY unidades DESC, provincia
    LIMIT 10
    """,
    [store_sales_urls, address_urls, item_urls],
).fetch_arrow_table()
print(pl.from_arrow(top_item_by_province))

customers_by_province: pa.Table = connection.execute(
    """
    SELECT
        trim(a.ca_county) AS provincia,
        count(*) AS clientes,
        count(DISTINCT a.ca_city) AS municipios_con_clientes,
        round(100.0 * count(*) / sum(count(*)) OVER (), 2) AS pct_clientes
    FROM read_parquet(?) AS c
    JOIN read_parquet(?) AS a
        ON c.c_current_addr_sk = a.ca_address_sk
    GROUP BY a.ca_county
    ORDER BY clientes DESC, provincia
    LIMIT 10
    """,
    [customer_urls, address_urls],
).fetch_arrow_table()
print(pl.from_arrow(customers_by_province))

store_buyers: pa.Table = connection.execute(
    """
    SELECT
        count(DISTINCT ss_customer_sk) AS clientes_con_compras,
        count(*) - count(ss_customer_sk) AS lineas_sin_cliente
    FROM read_parquet(?)
    """,
    [store_sales_urls],
).fetch_arrow_table()
print(pl.from_arrow(store_buyers))

sales_density: pa.Table = connection.execute(
    """
    WITH clientes AS (
        SELECT
            a.ca_state AS provincia_id,
            a.ca_county AS provincia,
            count(*) AS clientes
        FROM read_parquet(?) AS c
        JOIN read_parquet(?) AS a
            ON c.c_current_addr_sk = a.ca_address_sk
        GROUP BY a.ca_state, a.ca_county
    ),
    ventas AS (
        SELECT
            a.ca_state AS provincia_id,
            count(DISTINCT s.ss_ticket_number) AS tickets,
            sum(s.ss_ext_sales_price) AS ventas
        FROM read_parquet(?) AS s
        JOIN read_parquet(?) AS c
            ON s.ss_customer_sk = c.c_customer_sk
        JOIN read_parquet(?) AS a
            ON c.c_current_addr_sk = a.ca_address_sk
        GROUP BY a.ca_state
    )
    SELECT
        trim(c.provincia) AS provincia,
        c.clientes,
        v.tickets,
        round(100.0 * c.clientes / sum(c.clientes) OVER (), 2) AS pct_clientes,
        round(100.0 * v.ventas / sum(v.ventas) OVER (), 2) AS pct_ventas,
        round(v.ventas / c.clientes, 2) AS ventas_por_cliente
    FROM clientes AS c
    JOIN ventas AS v
        ON c.provincia_id = v.provincia_id
    ORDER BY v.ventas DESC, provincia
    LIMIT 10
    """,
    [customer_urls, address_urls, store_sales_urls, customer_urls, address_urls],
).fetch_arrow_table()
print(pl.from_arrow(sales_density))

connection.register("ine_municipios", ine_municipalities)

customers_per_municipality: pa.Table = connection.execute(
    """
    WITH clientes AS (
        SELECT
            trim(a.ca_state) AS provincia_id,
            count(*) AS clientes
        FROM read_parquet(?) AS c
        JOIN read_parquet(?) AS a
            ON c.c_current_addr_sk = a.ca_address_sk
        GROUP BY trim(a.ca_state)
    ),
    catalogo AS (
        SELECT provincia_id, provincia, count(*) AS municipios
        FROM ine_municipios
        GROUP BY provincia_id, provincia
    )
    SELECT
        m.provincia,
        m.municipios,
        c.clientes,
        round(c.clientes / m.municipios, 1) AS clientes_por_municipio
    FROM catalogo AS m
    JOIN clientes AS c
        ON m.provincia_id = c.provincia_id
    ORDER BY c.clientes DESC, m.provincia
    LIMIT 10
    """,
    [customer_urls, address_urls],
).fetch_arrow_table()
print(pl.from_arrow(customers_per_municipality))

plan: list[tuple[object, ...]] = connection.execute(
    """
    EXPLAIN
    SELECT d_year, d_holiday
    FROM read_parquet(?)
    WHERE d_year = 2000
    """,
    [date_urls],
).fetchall()
print(plan)

plan_select_all: list[tuple[object, ...]] = connection.execute(
    """
    EXPLAIN
    SELECT *
    FROM read_parquet(?)
    WHERE d_year = 2000
    """,
    [date_urls],
).fetchall()
print(plan_select_all)

docker exec -it trino-hdfs trino --server http://trino-hdfs:8080 --user luser

make -C entorno status

!curl --fail "http://namenode:9870/webhdfs/v1/datalake?op=GETFILESTATUS&user.name=luser"

make -C entorno warehouse-up

Dónde	Cómo aparece en este notebook	Para qué se usa
Terminal de tu equipo	Bloques de texto `bash` marcados «en una terminal de tu equipo»	`make`, `docker compose`, `docker exec`: arrancar, parar y borrar contenedores
Celdas de código de este notebook	Celdas ejecutables; las que empiezan por `!` lanzan una orden de shell	Se ejecutan dentro de `namenode` como `luser`: `hdfs dfs ...`, Python, consultas a Trino
Otro contenedor, desde tu equipo	`docker exec <contenedor> ...` en un bloque de texto	Entrar en un contenedor concreto, por ejemplo en el cliente de Trino

Capa	Qué guarda	Cómo se llega	Para qué sirve
`raw` (bronze)	Copia fiel de la fuente, en su formato e idioma original	Ingesta directa: copia o conversión de formato sin reinterpretar el contenido	Reproducibilidad: si cambia una regla de negocio, se reprocesa desde aquí sin volver a pedir el dato a la fuente
`silver`	Datos validados: tipos correctos, duplicados resueltos, claves de negocio unidas entre fuentes	Limpieza y `JOIN`: validar, tipar, deduplicar, enriquecer	Base común y reutilizable para varios consumidores, sin repetir la limpieza en cada consulta
`gold`	Agregados y modelos orientados a una pregunta de negocio concreta (por ejemplo, ventas por región y mes)	Agregación y modelado sobre `silver`, normalmente ya como tabla de hechos/dimensión	Consumo directo: cuadros de mando, informes, modelos de aprendizaje automático

Esquema de Trino	Volumen nominal sin comprimir	Parquet estimado	Con tres réplicas HDFS
`tiny`	0,01 GB	no extrapolable con fiabilidad	no extrapolable
`sf1`	1 GB	261 MiB medidos	783 MiB medidos
`sf10`	10 GB	2,5 GiB	7,6 GiB
`sf100`	100 GB	25 GiB	76 GiB
`sf300`	300 GB	76 GiB	229 GiB
`sf1000`	1 TB	255 GiB	765 GiB
`sf3000`	3 TB	0,75 TiB	2,2 TiB
`sf10000`	10 TB	2,5 TiB	7,5 TiB
`sf30000`	30 TB	7,5 TiB	22 TiB
`sf100000`	100 TB	25 TiB	75 TiB

Tabla	Filas aproximadas en SF1	Papel	Descripción breve
`call_center`	6	dimensión	Centros de atención telefónica y su organización.
`catalog_page`	11.718	dimensión	Páginas de los catálogos comerciales.
`catalog_returns`	144.067	hecho	Devoluciones de compras realizadas mediante catálogo.
`catalog_sales`	1.441.548	hecho	Líneas de pedidos realizados mediante catálogo.
`customer`	100.000	dimensión	Clientes y referencias a sus datos asociados.
`customer_address`	50.000	dimensión	Domicilios sintéticos de los clientes.
`customer_demographics`	1.920.800	dimensión	Características demográficas de los clientes.
`date_dim`	73.049	dimensión	Calendario con fechas y atributos temporales.
`household_demographics`	7.200	dimensión	Datos agregados del hogar y potencial de compra.
`income_band`	20	dimensión	Límites de los tramos de ingresos.
`inventory`	11.745.000	hecho de inventario	Existencias diarias por producto y almacén.
`item`	18.000	dimensión	Productos, precios, marcas y categorías.
`promotion`	300	dimensión	Promociones, descuentos y canales publicitarios.
`reason`	75	dimensión	Motivos utilizados en las devoluciones.
`ship_mode`	20	dimensión	Modos de transporte y sus operadores.
`store`	12	dimensión	Tiendas físicas, localización y organización.
`store_returns`	287.514	hecho	Devoluciones realizadas en tiendas.
`store_sales`	2.880.404	hecho	Líneas de ventas realizadas en tiendas.
`time_dim`	86.400	dimensión	Instantes del día y clasificación horaria.
`warehouse`	5	dimensión	Almacenes de distribución y localización.
`web_page`	60	dimensión	Páginas de los sitios web comerciales.
`web_returns`	71.763	hecho	Devoluciones de compras realizadas por web.
`web_sales`	719.384	hecho	Líneas de pedidos realizados por web.
`web_site`	30	dimensión	Sitios web, empresas y localización.

Columna	Significado
`municipio_id`	Código INE de cinco dígitos.
`provincia_id`	Sus dos primeros dígitos; identifica la provincia.
`codigo_municipio`	Código dentro de la provincia.
`digito_control`	Permite detectar errores de escritura.
`municipio`	Denominación oficial.
`provincia`	Denominación de la provincia en el libro del INE.

Columna	Tipo	Qué es
`d_date_sk`	`bigint`	Clave de la fecha; es la que referencian las tablas de hechos
`d_date`	`date`	La fecha del calendario
`d_year`	`integer`	El año
`d_holiday`	`char(1)`	Indicador de día festivo

Columna	Tabla	Qué es
`ws_item_sk`	`web_sales`	Producto vendido; con `ws_order_number` forma la clave primaria
`ws_order_number`	`web_sales`	Número de pedido
`ws_bill_customer_sk`	`web_sales`	Cliente al que se factura; apunta a `customer.c_customer_sk`
`c_customer_sk`	`customer`	Clave del cliente

Tabla	Prefijo	Una fila es…	Columnas para recordar
`customer`	`c_`	un cliente	`c_customer_sk`, `c_first_name`, `c_last_name`, `c_current_addr_sk`
`customer_address`	`ca_`	un domicilio	`ca_address_sk`, `ca_city` (municipio), `ca_county` (provincia), `ca_state` (código de provincia)
`item`	`i_`	un producto	`i_item_sk`, `i_product_name`, `i_category`, `i_current_price`
`date_dim`	`d_`	un día del calendario	`d_date_sk`, `d_date`, `d_year`, `d_moy`
`time_dim`	`t_`	un segundo del día	`t_time_sk`, `t_hour`, `t_minute`
`store`	`s_`	una tienda física	`s_store_sk`, `s_store_name`, `s_city`
`web_sales`	`ws_`	una línea de un pedido web	`ws_order_number`, `ws_item_sk`, `ws_bill_customer_sk`, `ws_sold_date_sk`, `ws_net_profit`
`store_sales`	`ss_`	una línea de un ticket de tienda	`ss_ticket_number`, `ss_item_sk`, `ss_customer_sk`, `ss_store_sk`, `ss_net_profit`

Pregunta	Hecho	Dimensión	Clave de unión	Resultado que se estudia
¿Qué cliente ha realizado más pedidos?	`web_sales`	`customer`	`ws_bill_customer_sk = c_customer_sk`	Pedidos distintos por cliente
¿De qué provincia proceden más compras?	`web_sales`	`customer_address`	`ws_bill_addr_sk = ca_address_sk`	Pedidos distintos por provincia (`ca_county`)
¿Cómo se distribuyen las compras por hora?	`web_sales`	`time_dim`	`ws_sold_time_sk = t_time_sk`	Pedidos distintos por `t_hour`

Columna	Tabla	Qué es
`ss_sold_date_sk`	`store_sales`	Fecha de la venta; apunta a `date_dim.d_date_sk`
`ss_item_sk`	`store_sales`	Producto vendido; apunta a `item.i_item_sk`
`ss_ext_sales_price`	`store_sales`	Importe de la línea de venta
`i_category`	`item`	Categoría del producto
`d_year`	`date_dim`	Año de la fecha

Pregunta	Dimensiones	Clave de unión	Se agrupa por
¿Cuánto se vende en cada canal?	ninguna	—	el propio hecho: tienda, catálogo o web
¿Qué categorías y qué productos se venden más?	`item`	`ss_item_sk = i_item_sk`	`i_category`; después `i_item_sk`
¿Cómo se reparten las ventas por mes?	`date_dim`	`ss_sold_date_sk = d_date_sk`	`d_moy`
¿Qué producto se vende más cada mes?	`date_dim`, `item`	las dos anteriores	`d_moy` y producto
¿Qué producto se vende más en cada provincia?	`customer_address`, `item`	`ss_addr_sk = ca_address_sk`	`ca_county` y producto
¿Dónde viven los clientes?	`customer_address`, desde `customer`	`c_current_addr_sk = ca_address_sk`	`ca_county`
¿Dónde se concentran las ventas?	`customer`, `customer_address`	`ss_customer_sk = c_customer_sk`	`ca_county`

Canal	Hecho	Prefijo	Identifica la venta	Dimensiones propias del canal
Tienda física	`store_sales`	`ss_`	`ss_ticket_number`	`store` (`ss_store_sk`)
Catálogo	`catalog_sales`	`cs_`	`cs_order_number`	`call_center` (`cs_call_center_sk`) y `catalog_page` (`cs_catalog_page_sk`)
Web	`web_sales`	`ws_`	`ws_order_number`	`web_site` (`ws_web_site_sk`) y `web_page` (`ws_web_page_sk`)

Orden	Qué hace	Qué se conserva	Qué se pierde
`make -C entorno stop`	Para los contenedores sin eliminarlos (`docker compose stop`)	Todo: HDFS, los metadatos del Hive Metastore y también el interior de cada contenedor (ficheros del directorio de trabajo del kernel, `/tmp` de `namenode`, histórico del Timeline Server)	Sólo los procesos en marcha: Jupyter y las variables de los kernels
`make -C entorno down`	Elimina los contenedores (`docker compose down`)	Los volúmenes Docker: HDFS (`/datalake`, `/warehouse`, `/user`) y los metadatos del Hive Metastore	Lo que vivía dentro de los contenedores: ficheros locales de `namenode` e histórico del Timeline Server
`make -C entorno clean`	`down` y, además, borra el volumen de PostgreSQL (Hive Metastore) y el del almacenamiento de objetos de la sesión 3	Todo HDFS: el dataset TPC-DS/INE de esta sesión sigue disponible	Las definiciones de tablas del metastore y los objetos de la sesión 3
`make -C entorno clean-hdfs`	`clean` y, además, borra los volúmenes HDFS de `namenode` y de los tres DataNodes	Sólo las imágenes ya descargadas	Todos los datos, incluido el trabajo de la Sesión 1: el siguiente arranque parte de un HDFS recién formateado

Columna	Tipo	Descripción	Clave / relación
`d_date_sk`	`bigint`	Clave surrogate que identifica cada fecha del calendario.	PK
`d_date_id`	`char(16)`	Identificador de negocio de la fecha.	BK
`d_date`	`date`	Fecha del calendario.	—
`d_month_seq`	`integer`	Secuencia del mes en el calendario.	—
`d_week_seq`	`integer`	Secuencia de la semana en el calendario.	—
`d_quarter_seq`	`integer`	Secuencia del trimestre en el calendario.	—
`d_year`	`integer`	Año del calendario.	—
`d_dow`	`integer`	Día de la semana, codificado como número.	—
`d_moy`	`integer`	Mes del año, codificado como número.	—
`d_dom`	`integer`	Día del mes, codificado como número.	—
`d_qoy`	`integer`	Trimestre del año, codificado como número.	—
`d_fy_year`	`integer`	Año fiscal del calendario.	—
`d_fy_quarter_seq`	`integer`	Número secuencial del trimestre fiscal dentro del calendario.	—
`d_fy_week_seq`	`integer`	Número secuencial de la semana fiscal dentro del calendario.	—
`d_day_name`	`char(9)`	Nombre del día de la semana.	—
`d_quarter_name`	`char(6)`	Nombre del trimestre.	—
`d_holiday`	`char(1)`	Indicador de día festivo.	—
`d_weekend`	`char(1)`	Indicador de fin de semana.	—
`d_following_holiday`	`char(1)`	Indicador de día anterior a un festivo.	—
`d_first_dom`	`integer`	Primer día del mes.	—
`d_last_dom`	`integer`	Último día del mes.	—
`d_same_day_ly`	`integer`	Clave del mismo día del año anterior.	—
`d_same_day_lq`	`integer`	Clave del mismo día del trimestre anterior.	—
`d_current_day`	`char(1)`	Indicador de que es el día actual.	—
`d_current_week`	`char(1)`	Indicador de que pertenece a la semana actual.	—
`d_current_month`	`char(1)`	Indicador de que pertenece al mes actual.	—
`d_current_quarter`	`char(1)`	Indicador de que pertenece al trimestre actual.	—
`d_current_year`	`char(1)`	Indicador de que pertenece al año actual.	—

Columna	Tipo	Descripción	Clave / relación
`t_time_sk`	`bigint`	Clave surrogate que identifica cada segundo del día.	PK
`t_time_id`	`char(16)`	Identificador de negocio de hora.	BK
`t_time`	`integer`	Instante del día expresado como segundos desde medianoche.	—
`t_hour`	`integer`	Hora del día.	—
`t_minute`	`integer`	Minuto de la hora.	—
`t_second`	`integer`	Segundo del minuto.	—
`t_am_pm`	`char(2)`	Indicador de mañana o tarde.	—
`t_shift`	`char(20)`	Turno del establecimiento.	—
`t_sub_shift`	`char(20)`	Subturno dentro del turno.	—
`t_meal_time`	`char(20)`	Franja asociada a una comida.	—

Columna	Tipo	Descripción	Clave / relación
`c_customer_sk`	`bigint`	Clave surrogate que identifica el cliente.	PK
`c_customer_id`	`char(16)`	Identificador de negocio de cliente.	BK
`c_current_cdemo_sk`	`bigint`	Perfil demográfico actual del cliente.	FK → customer_demographics.cd_demo_sk
`c_current_hdemo_sk`	`bigint`	Perfil actual del hogar del cliente.	FK → household_demographics.hd_demo_sk
`c_current_addr_sk`	`bigint`	Dirección actual del cliente.	FK → customer_address.ca_address_sk
`c_first_shipto_date_sk`	`bigint`	Fecha del primer envío al cliente.	FK → date_dim.d_date_sk
`c_first_sales_date_sk`	`bigint`	Fecha de la primera compra del cliente.	FK → date_dim.d_date_sk
`c_salutation`	`char(10)`	Tratamiento o saludo del cliente.	—
`c_first_name`	`char(20)`	Nombre de pila del cliente.	—
`c_last_name`	`char(30)`	Apellidos del cliente.	—
`c_preferred_cust_flag`	`char(1)`	Indicador de cliente preferente.	—
`c_birth_day`	`integer`	Día de nacimiento.	—
`c_birth_month`	`integer`	Mes de nacimiento.	—
`c_birth_year`	`integer`	Año de nacimiento.	—
`c_birth_country`	`varchar(20)`	País de nacimiento.	—
`c_login`	`char(13)`	Identificador de acceso del cliente.	—
`c_email_address`	`char(50)`	Dirección de correo electrónico del cliente.	—
`c_last_review_date_sk`	`bigint`	Fecha de la última revisión de los datos del cliente.	FK → date_dim.d_date_sk

Columna	Tipo	Descripción	Clave / relación
`s_store_sk`	`bigint`	Clave surrogate que identifica la tienda.	PK
`s_store_id`	`char(16)`	Identificador de negocio de tienda.	BK
`s_rec_start_date`	`date`	Fecha de inicio de vigencia de esta versión del registro.	—
`s_rec_end_date`	`date`	Fecha de fin de vigencia de esta versión del registro; vacía en la versión vigente.	—
`s_closed_date_sk`	`bigint`	Fecha de cierre de la tienda.	FK → date_dim.d_date_sk
`s_store_name`	`varchar(50)`	Nombre de la tienda.	—
`s_number_employees`	`integer`	Número de empleados de la tienda.	—
`s_floor_space`	`integer`	Superficie de la tienda.	—
`s_hours`	`char(20)`	Horario de apertura de la tienda.	—
`s_manager`	`varchar(40)`	Nombre del responsable de la tienda.	—
`s_market_id`	`integer`	Identificador del mercado al que pertenece la tienda.	—
`s_geography_class`	`varchar(100)`	Clase geográfica de la zona de la tienda.	—
`s_market_desc`	`varchar(100)`	Descripción del mercado.	—
`s_market_manager`	`varchar(40)`	Nombre del responsable de mercado.	—
`s_division_id`	`integer`	Identificador de la división.	—
`s_division_name`	`varchar(50)`	Nombre de la división.	—
`s_company_id`	`integer`	Identificador de la empresa.	—
`s_company_name`	`varchar(50)`	Nombre de la empresa.	—
`s_street_number`	`varchar(10)`	Número de la vía.	—
`s_street_name`	`varchar(60)`	Nombre de la vía.	—
`s_street_type`	`char(15)`	Tipo de vía.	—
`s_suite_number`	`varchar(10)`	Número de local o suite.	—
`s_city`	`varchar(60)`	Municipio asignado desde el INE.	—
`s_county`	`varchar(30)`	Provincia según el INE.	—
`s_state`	`char(2)`	Código INE de provincia.	—
`s_zip`	`char(10)`	Código postal.	—
`s_country`	`varchar(20)`	País fijado a `España` tras el cruce.	—
`s_gmt_offset`	`decimal(5,2)`	Desplazamiento horario respecto de GMT.	—
`s_tax_precentage`	`decimal(5,2)`	Porcentaje de impuesto aplicado en la tienda.	—

Columna	Tipo	Descripción	Clave / relación
`cc_call_center_sk`	`bigint`	Clave surrogate que identifica el centro de llamadas.	PK
`cc_call_center_id`	`char(16)`	Identificador de negocio de centro de llamadas.	BK
`cc_rec_start_date`	`date`	Fecha de inicio de vigencia de esta versión del registro.	—
`cc_rec_end_date`	`date`	Fecha de fin de vigencia de esta versión del registro; vacía en la versión vigente.	—
`cc_closed_date_sk`	`integer`	Fecha de cierre del centro.	FK → date_dim.d_date_sk
`cc_open_date_sk`	`integer`	Fecha de apertura del centro.	FK → date_dim.d_date_sk
`cc_name`	`varchar(50)`	Nombre del centro de llamadas.	—
`cc_class`	`varchar(50)`	Clase del centro de atención telefónica.	—
`cc_employees`	`integer`	Número de empleados del centro.	—
`cc_sq_ft`	`integer`	Superficie del centro en pies cuadrados.	—
`cc_hours`	`char(20)`	Horario de apertura del centro.	—
`cc_manager`	`varchar(40)`	Nombre del responsable del centro.	—
`cc_mkt_id`	`integer`	Identificador del mercado.	—
`cc_mkt_class`	`char(50)`	Descripción de la clase de mercado del centro.	—
`cc_mkt_desc`	`varchar(100)`	Descripción del mercado.	—
`cc_market_manager`	`varchar(40)`	Nombre del responsable de mercado del centro.	—
`cc_division`	`integer`	División de la empresa a la que pertenece el centro.	—
`cc_division_name`	`varchar(50)`	Nombre de la división.	—
`cc_company`	`integer`	Identificador de la empresa que opera el centro.	—
`cc_company_name`	`char(50)`	Nombre de la empresa.	—
`cc_street_number`	`char(10)`	Número de la vía.	—
`cc_street_name`	`varchar(60)`	Nombre de la vía.	—
`cc_street_type`	`char(15)`	Tipo de vía de la dirección del centro.	—
`cc_suite_number`	`char(10)`	Número de local o suite.	—
`cc_city`	`varchar(60)`	Municipio asignado desde el INE.	—
`cc_county`	`varchar(30)`	Provincia según el INE.	—
`cc_state`	`char(2)`	Código INE de provincia.	—
`cc_zip`	`char(10)`	Código postal del centro.	—
`cc_country`	`varchar(20)`	País fijado a `España` tras el cruce.	—
`cc_gmt_offset`	`decimal(5,2)`	Desplazamiento horario del centro respecto de GMT.	—
`cc_tax_percentage`	`decimal(5,2)`	Porcentaje de impuesto aplicado por el centro.	—

Columna	Tipo	Descripción	Clave / relación
`sm_ship_mode_sk`	`bigint`	Clave surrogate que identifica el modo de envío.	PK
`sm_ship_mode_id`	`char(16)`	Identificador de negocio del modo de envío.	BK
`sm_type`	`char(30)`	Tipo de modo de envío.	—
`sm_code`	`char(10)`	Código del modo de envío.	—
`sm_carrier`	`char(20)`	Transportista responsable del modo de envío.	—
`sm_contract`	`char(20)`	Contrato asociado al modo de envío.	—

Columna	Tipo	Descripción	Clave / relación
`i_item_sk`	`bigint`	Clave surrogate que identifica el producto.	PK
`i_item_id`	`char(16)`	Identificador de negocio de producto.	BK
`i_rec_start_date`	`date`	Fecha de inicio de vigencia de esta versión del registro.	—
`i_rec_end_date`	`date`	Fecha de fin de vigencia de esta versión del registro; vacía en la versión vigente.	—
`i_item_desc`	`varchar(200)`	Descripción textual del producto.	—
`i_current_price`	`decimal(7,2)`	Precio minorista vigente del producto.	—
`i_wholesale_cost`	`decimal(7,2)`	Coste mayorista del producto.	—
`i_brand_id`	`integer`	Identificador numérico de la marca.	—
`i_brand`	`char(50)`	Marca del producto.	—
`i_class_id`	`integer`	Identificador numérico de la clase.	—
`i_class`	`char(50)`	Clase del producto dentro de su categoría.	—
`i_category_id`	`integer`	Identificador numérico de la categoría.	—
`i_category`	`char(50)`	Categoría del producto.	—
`i_manufact_id`	`integer`	Identificador numérico del fabricante.	—
`i_manufact`	`char(50)`	Nombre del fabricante del producto.	—
`i_size`	`char(20)`	Tamaño del producto.	—
`i_formulation`	`char(20)`	Código de formulación del producto.	—
`i_color`	`char(20)`	Color del producto.	—
`i_units`	`char(10)`	Unidad de medida del producto.	—
`i_container`	`char(10)`	Tipo de envase del producto.	—
`i_manager_id`	`integer`	Identificador del responsable del producto.	—
`i_product_name`	`char(50)`	Nombre comercial del producto.	—

Columna	Tipo	Descripción	Clave / relación
`ca_address_sk`	`bigint`	Clave surrogate que identifica la dirección del cliente.	PK
`ca_address_id`	`char(16)`	Identificador de negocio de dirección.	BK
`ca_street_number`	`char(10)`	Número de la vía.	—
`ca_street_name`	`char(60)`	Nombre de la vía.	—
`ca_street_type`	`char(15)`	Tipo de vía.	—
`ca_suite_number`	`char(10)`	Número de apartamento o suite.	—
`ca_city`	`char(60)`	Municipio asignado desde el INE.	—
`ca_county`	`char(30)`	Provincia según el INE.	—
`ca_state`	`char(2)`	Código INE de provincia.	—
`ca_zip`	`char(10)`	Código postal.	—
`ca_country`	`char(20)`	País fijado a `España` tras el cruce.	—
`ca_gmt_offset`	`decimal(5,2)`	Desplazamiento horario respecto de GMT.	—
`ca_location_type`	`char(20)`	Tipo de ubicación.	—

Columna	Tipo	Descripción	Clave / relación
`cd_demo_sk`	`bigint`	Clave surrogate que identifica la demografía del cliente.	PK
`cd_gender`	`char(1)`	Género.	—
`cd_marital_status`	`char(1)`	Estado civil.	—
`cd_education_status`	`char(20)`	Nivel educativo.	—
`cd_purchase_estimate`	`integer`	Estimación del gasto en compras.	—
`cd_credit_rating`	`char(10)`	Calificación crediticia.	—
`cd_dep_count`	`integer`	Número de personas dependientes.	—
`cd_dep_employed_count`	`integer`	Número de dependientes con empleo.	—
`cd_dep_college_count`	`integer`	Número de dependientes que cursan estudios universitarios.	—

Columna	Tipo	Descripción	Clave / relación
`hd_demo_sk`	`bigint`	Clave surrogate que identifica la demografía del hogar.	PK
`hd_income_band_sk`	`bigint`	Tramo de ingresos del hogar.	FK → income_band.ib_income_band_sk
`hd_buy_potential`	`char(15)`	Potencial estimado de compra del hogar.	—
`hd_dep_count`	`integer`	Número de personas dependientes en el hogar.	—
`hd_vehicle_count`	`integer`	Número de vehículos del hogar.	—

Columna	Tipo	Descripción	Clave / relación
`ib_income_band_sk`	`bigint`	Clave surrogate que identifica el tramo de ingresos.	PK
`ib_lower_bound`	`integer`	Límite inferior del tramo de ingresos.	—
`ib_upper_bound`	`integer`	Límite superior del tramo de ingresos.	—

Columna	Tipo	Descripción	Clave / relación
`w_warehouse_sk`	`bigint`	Clave surrogate que identifica el almacén.	PK
`w_warehouse_id`	`char(16)`	Identificador de negocio de almacén.	BK
`w_warehouse_name`	`varchar(20)`	Nombre del almacén.	—
`w_warehouse_sq_ft`	`integer`	Superficie del almacén en pies cuadrados.	—
`w_street_number`	`char(10)`	Número de la vía.	—
`w_street_name`	`char(60)`	Nombre de la vía.	—
`w_street_type`	`char(15)`	Tipo de vía de la dirección del almacén.	—
`w_suite_number`	`char(10)`	Número de local o suite.	—
`w_city`	`char(60)`	Municipio asignado desde el INE.	—
`w_county`	`char(30)`	Provincia según el INE.	—
`w_state`	`char(2)`	Código INE de provincia.	—
`w_zip`	`char(10)`	Código postal del almacén.	—
`w_country`	`char(20)`	País fijado a `España` tras el cruce.	—
`w_gmt_offset`	`decimal(5,2)`	Desplazamiento horario del almacén respecto de GMT.	—

Columna	Tipo	Descripción	Clave / relación
`cp_catalog_page_sk`	`bigint`	Clave surrogate que identifica la página del catálogo.	PK
`cp_catalog_page_id`	`char(16)`	Identificador de negocio de página de catálogo.	BK
`cp_start_date_sk`	`integer`	Fecha de inicio de vigencia de la página.	FK → date_dim.d_date_sk
`cp_end_date_sk`	`integer`	Fecha de fin de vigencia de la página.	FK → date_dim.d_date_sk
`cp_department`	`varchar(50)`	Departamento al que pertenece la página.	—
`cp_catalog_number`	`integer`	Número del catálogo.	—
`cp_catalog_page_number`	`integer`	Número de la página dentro del catálogo.	—
`cp_description`	`varchar(100)`	Descripción de la página.	—
`cp_type`	`varchar(100)`	Tipo de página.	—

Columna	Tipo	Descripción	Clave / relación
`p_promo_sk`	`bigint`	Clave surrogate que identifica la promoción.	PK
`p_promo_id`	`char(16)`	Identificador de negocio de promoción.	BK
`p_start_date_sk`	`bigint`	Fecha de inicio de la promoción.	FK → date_dim.d_date_sk
`p_end_date_sk`	`bigint`	Fecha de fin de la promoción.	FK → date_dim.d_date_sk
`p_item_sk`	`bigint`	Producto promocionado.	FK → item.i_item_sk
`p_cost`	`decimal(15,2)`	Coste de la promoción.	—
`p_response_targe`	`integer`	Objetivo de respuesta de la promoción.	—
`p_promo_name`	`char(50)`	Nombre de la promoción.	—
`p_channel_dmail`	`char(1)`	Indicador de uso del canal de correo postal.	—
`p_channel_email`	`char(1)`	Indicador de uso del canal de correo electrónico.	—
`p_channel_catalog`	`char(1)`	Indicador de uso del canal de catálogo.	—
`p_channel_tv`	`char(1)`	Indicador de uso del canal de televisión.	—
`p_channel_radio`	`char(1)`	Indicador de uso del canal de radio.	—
`p_channel_press`	`char(1)`	Indicador de uso del canal de prensa.	—
`p_channel_event`	`char(1)`	Indicador de uso del canal de eventos.	—
`p_channel_demo`	`char(1)`	Indicador de uso del canal de demostraciones.	—
`p_channel_details`	`varchar(100)`	Detalles de los canales utilizados.	—
`p_purpose`	`char(15)`	Finalidad de la promoción.	—
`p_discount_active`	`char(1)`	Indicador de descuento activo.	—

Columna	Tipo	Descripción	Clave / relación
`r_reason_sk`	`bigint`	Clave surrogate que identifica el motivo.	PK
`r_reason_id`	`char(16)`	Identificador de negocio del motivo.	BK
`r_reason_desc`	`char(100)`	Descripción del motivo.	—

Columna	Tipo	Descripción	Clave / relación
`wp_web_page_sk`	`bigint`	Clave surrogate que identifica la página web.	PK
`wp_web_page_id`	`char(16)`	Identificador de negocio de la página web.	BK
`wp_rec_start_date`	`date`	Fecha de inicio de vigencia de esta versión del registro.	—
`wp_rec_end_date`	`date`	Fecha de fin de vigencia de esta versión del registro; vacía en la versión vigente.	—
`wp_creation_date_sk`	`integer`	Fecha de creación de la página.	FK → date_dim.d_date_sk
`wp_access_date_sk`	`integer`	Fecha del último acceso a la página.	FK → date_dim.d_date_sk
`wp_autogen_flag`	`char(1)`	Indicador de página generada automáticamente.	—
`wp_customer_sk`	`bigint`	Cliente al que está asociada la página.	FK → customer.c_customer_sk
`wp_url`	`varchar(100)`	URL de la página.	—
`wp_type`	`char(50)`	Tipo de página.	—
`wp_char_count`	`integer`	Número de caracteres de la página.	—
`wp_link_count`	`integer`	Número de enlaces de la página.	—
`wp_image_count`	`integer`	Número de imágenes de la página.	—
`wp_max_ad_count`	`integer`	Número máximo de anuncios de la página.	—

Sesión 2 — Exploración de TPC-DS en Parquet sobre HDFS¶

Objetivos¶

Antes de empezar: Jupyter en namenode¶

Dónde se ejecuta cada orden¶

Arrancar el clúster¶

Arrancar Jupyter dentro de namenode¶

Conectar este notebook con Jupyter¶

Diapositivas de la sesión¶

Sesión 2: TPC-DS en Parquet sobre HDFS

Cómo vamos a recorrer la sesión

Cómo se evalúa esta sesión

Qué es un data lake

Data lakes y arquitectura medallion¶

Para profundizar¶

Qué vamos a construir y con qué datos

Qué es Trino¶

TPC-DS y la escala SF1

TPC-DS y la escala SF1¶

La ruta de los datos en esta sesión

Arquitectura de la sesión¶

Dónde vive cada cosa

Arrancar los servicios del warehouse

Arrancar los servicios del warehouse¶

Generamos el dataset, paso a paso

Generar los Parquet de TPC-DS¶

Preparar Python¶

La segunda fuente, los municipios del INE

Incorporar la relación de municipios del INE¶

Publicar la fuente del INE en HDFS¶

Conectar con Trino desde el notebook¶

Almacenado temporal del CSV en Hive Metastore¶

Unir las dos fuentes de forma determinista¶

Construir las consultas de localización¶

Materializar las 24 tablas¶

Un bucle que se puede repetir

Validar y retirar sólo los metadatos temporales¶

Cómo se preparó la instantánea del INE (referencia)

Descargar, limpiar y convertir la fuente (referencia)¶

Descargar la publicación¶

Leer las hojas con Polars¶

Normalizar los municipios¶

Validar la relación¶

Convertir y comprimir¶

Dónde están físicamente los datos¶

Los datos ya están en HDFS

El modelo de datos y la primera lectura

Qué tablas contiene TPC-DS¶

Tres formas de leer los mismos ficheros

Otras formas de acceder a HDFS¶

Acceso con WebHDFS y fsspec¶

Leer la misma tabla con tres bibliotecas¶

Ya hemos leído Parquet con Python

Explorar con PyArrow¶

Explorar con Polars¶

Explorar con DuckDB¶

De ficheros a relaciones

Ver la tabla resultante y comprobar relaciones¶

Declarar una copia con clave ajena en DuckDB¶

Las tablas con datos delante¶

Tres bibliotecas, un mismo dato

Preguntas de negocio de verdad

Preguntas sencillas de negocio sobre las ventas web¶

Cliente con más pedidos: PyArrow¶

Provincia con más compras: Polars¶

Comprobar la procedencia de los lugares¶

Distribución horaria de las compras: DuckDB¶

Recuperar el identificador del cliente con DuckDB¶

Actividades de los casos de uso¶

Tres preguntas, dos fuentes

Una consulta con tres tablas a la vez

Una consulta que combina varias tablas¶

Las primeras preguntas ante un dataset de ventas

Las primeras preguntas ante un dataset de ventas¶

Cuánto se vende en cada canal¶

Qué categorías y qué productos se venden más¶

Cómo se reparten las ventas por mes¶

Qué producto se vende más cada mes¶

Qué producto se vende más en cada provincia¶

Dónde viven los clientes¶

Dónde se concentran las ventas¶

Antes de empezar: Jupyter en `namenode`¶

Arrancar Jupyter dentro de `namenode`¶

`date_dim`¶

`time_dim`¶

`item`¶

`customer`¶

`customer_address`¶

`customer_demographics`¶

`household_demographics`¶

`income_band`¶

`store`¶

`warehouse`¶

`call_center`¶

`catalog_page`¶

`promotion`¶

`reason`¶

`ship_mode`¶

`web_page`¶

`web_site`¶

`store_sales`¶

`catalog_sales`¶

`web_sales`¶

`store_returns`¶

`catalog_returns`¶

`web_returns`¶

`inventory`¶

Columna	Tipo	Descripción	Clave / relación
`web_site_sk`	`bigint`	Clave surrogate que identifica el sitio web.	PK
`web_site_id`	`char(16)`	Identificador de negocio del sitio web.	BK
`web_rec_start_date`	`date`	Fecha de inicio de vigencia de esta versión del registro.	—
`web_rec_end_date`	`date`	Fecha de fin de vigencia de esta versión del registro; vacía en la versión vigente.	—
`web_name`	`char(50)`	Nombre del sitio web.	—
`web_open_date_sk`	`bigint`	Fecha de apertura del sitio web.	FK → date_dim.d_date_sk
`web_close_date_sk`	`bigint`	Fecha de cierre del sitio web.	FK → date_dim.d_date_sk
`web_class`	`varchar(50)`	Clase del sitio web.	—
`web_manager`	`varchar(40)`	Nombre del responsable del sitio web.	—
`web_mkt_id`	`integer`	Identificador del mercado.	—
`web_mkt_class`	`varchar(50)`	Descripción de la clase de mercado del sitio web.	—
`web_mkt_desc`	`varchar(100)`	Descripción del mercado.	—
`web_market_manager`	`varchar(40)`	Nombre del responsable de mercado del sitio web.	—
`web_company_id`	`integer`	Identificador de la empresa.	—
`web_company_name`	`varchar(50)`	Nombre de la empresa.	—
`web_street_number`	`varchar(10)`	Número de la vía.	—
`web_street_name`	`varchar(60)`	Nombre de la vía.	—
`web_street_type`	`char(15)`	Tipo de vía de la dirección del sitio web.	—
`web_suite_number`	`char(10)`	Número de local o suite.	—
`web_city`	`varchar(60)`	Municipio asignado desde el INE.	—
`web_county`	`varchar(30)`	Provincia según el INE.	—
`web_state`	`char(2)`	Código INE de provincia.	—
`web_zip`	`char(10)`	Código postal del sitio web.	—
`web_country`	`varchar(20)`	País fijado a `España` tras el cruce.	—
`web_gmt_offset`	`decimal(5,2)`	Desplazamiento horario del sitio respecto de GMT.	—
`web_tax_percentage`	`decimal(5,2)`	Porcentaje de impuesto aplicado por el sitio web.	—

Columna	Tipo	Descripción	Clave / relación
`ss_sold_date_sk`	`bigint`	Fecha de la venta.	FK → date_dim.d_date_sk
`ss_sold_time_sk`	`bigint`	Hora de la venta.	FK → time_dim.t_time_sk
`ss_item_sk`	`bigint`	Producto vendido; con `ss_ticket_number` forma la clave primaria.	PK (1/2); FK → item.i_item_sk
`ss_customer_sk`	`bigint`	Cliente que hace la compra.	FK → customer.c_customer_sk
`ss_cdemo_sk`	`bigint`	Perfil demográfico del cliente.	FK → customer_demographics.cd_demo_sk
`ss_hdemo_sk`	`bigint`	Perfil del hogar del cliente.	FK → household_demographics.hd_demo_sk
`ss_addr_sk`	`bigint`	Dirección del cliente.	FK → customer_address.ca_address_sk
`ss_store_sk`	`bigint`	Tienda en la que se hace la venta.	FK → store.s_store_sk
`ss_promo_sk`	`bigint`	Promoción aplicada a la venta.	FK → promotion.p_promo_sk
`ss_ticket_number`	`bigint`	Número de ticket; con `ss_item_sk` forma la clave primaria.	PK (2/2)
`ss_quantity`	`integer`	Cantidad de unidades vendidas.	—
`ss_wholesale_cost`	`decimal(7,2)`	Coste mayorista por unidad.	—
`ss_list_price`	`decimal(7,2)`	Precio de catálogo por unidad.	—
`ss_sales_price`	`decimal(7,2)`	Precio de venta por unidad.	—
`ss_ext_discount_amt`	`decimal(7,2)`	Importe extendido del descuento aplicado.	—
`ss_ext_sales_price`	`decimal(7,2)`	Importe extendido de la venta antes de descuentos e impuestos.	—
`ss_ext_wholesale_cost`	`decimal(7,2)`	Coste mayorista extendido de los productos vendidos.	—
`ss_ext_list_price`	`decimal(7,2)`	Precio de lista extendido de los productos vendidos.	—
`ss_ext_tax`	`decimal(7,2)`	Impuesto extendido de la venta.	—
`ss_coupon_amt`	`decimal(7,2)`	Importe total de los cupones aplicados.	—
`ss_net_paid`	`decimal(7,2)`	Importe neto pagado por la venta.	—
`ss_net_paid_inc_tax`	`decimal(7,2)`	Importe neto pagado incluyendo impuestos.	—
`ss_net_profit`	`decimal(7,2)`	Beneficio neto de la venta.	—

Columna	Tipo	Descripción	Clave / relación
`cs_sold_date_sk`	`bigint`	Fecha de la venta.	FK → date_dim.d_date_sk
`cs_sold_time_sk`	`bigint`	Hora de la venta.	FK → time_dim.t_time_sk
`cs_ship_date_sk`	`bigint`	Fecha de envío del pedido.	FK → date_dim.d_date_sk
`cs_bill_customer_sk`	`bigint`	Cliente al que se factura el pedido.	FK → customer.c_customer_sk
`cs_bill_cdemo_sk`	`bigint`	Perfil demográfico del cliente al que se factura.	FK → customer_demographics.cd_demo_sk
`cs_bill_hdemo_sk`	`bigint`	Perfil del hogar del cliente al que se factura.	FK → household_demographics.hd_demo_sk
`cs_bill_addr_sk`	`bigint`	Dirección de facturación.	FK → customer_address.ca_address_sk
`cs_ship_customer_sk`	`bigint`	Cliente que recibe el envío.	FK → customer.c_customer_sk
`cs_ship_cdemo_sk`	`bigint`	Perfil demográfico del cliente que recibe el envío.	FK → customer_demographics.cd_demo_sk
`cs_ship_hdemo_sk`	`bigint`	Perfil del hogar del cliente que recibe el envío.	FK → household_demographics.hd_demo_sk
`cs_ship_addr_sk`	`bigint`	Dirección de envío.	FK → customer_address.ca_address_sk
`cs_call_center_sk`	`bigint`	Centro de llamadas que tramitó el pedido.	FK → call_center.cc_call_center_sk
`cs_catalog_page_sk`	`bigint`	Página del catálogo desde la que se hizo el pedido.	FK → catalog_page.cp_catalog_page_sk
`cs_ship_mode_sk`	`bigint`	Modo de envío del pedido.	FK → ship_mode.sm_ship_mode_sk
`cs_warehouse_sk`	`bigint`	Almacén desde el que se sirve el pedido.	FK → warehouse.w_warehouse_sk
`cs_item_sk`	`bigint`	Producto vendido; con `cs_order_number` forma la clave primaria.	PK (1/2); FK → item.i_item_sk
`cs_promo_sk`	`bigint`	Promoción aplicada a la venta.	FK → promotion.p_promo_sk
`cs_order_number`	`bigint`	Número de pedido; con `cs_item_sk` forma la clave primaria.	PK (2/2)
`cs_quantity`	`integer`	Cantidad de unidades vendidas.	—
`cs_wholesale_cost`	`decimal(7,2)`	Coste mayorista por unidad.	—
`cs_list_price`	`decimal(7,2)`	Precio de catálogo por unidad.	—
`cs_sales_price`	`decimal(7,2)`	Precio de venta por unidad.	—
`cs_ext_discount_amt`	`decimal(7,2)`	Importe extendido del descuento aplicado.	—
`cs_ext_sales_price`	`decimal(7,2)`	Importe extendido de la venta antes de descuentos e impuestos.	—
`cs_ext_wholesale_cost`	`decimal(7,2)`	Coste mayorista extendido de los productos vendidos.	—
`cs_ext_list_price`	`decimal(7,2)`	Precio de lista extendido de los productos vendidos.	—
`cs_ext_tax`	`decimal(7,2)`	Impuesto extendido de la venta.	—
`cs_coupon_amt`	`decimal(7,2)`	Importe total de los cupones aplicados.	—
`cs_ext_ship_cost`	`decimal(7,2)`	Coste extendido del envío del pedido.	—
`cs_net_paid`	`decimal(7,2)`	Importe neto pagado por la venta.	—
`cs_net_paid_inc_tax`	`decimal(7,2)`	Importe neto pagado incluyendo impuestos.	—
`cs_net_paid_inc_ship`	`decimal(7,2)`	Importe neto pagado incluyendo el envío.	—
`cs_net_paid_inc_ship_tax`	`decimal(7,2)`	Importe neto pagado incluyendo envío e impuestos.	—
`cs_net_profit`	`decimal(7,2)`	Beneficio neto de la venta.	—

Columna	Tipo	Descripción	Clave / relación
`ws_sold_date_sk`	`bigint`	Fecha de la venta.	FK → date_dim.d_date_sk
`ws_sold_time_sk`	`bigint`	Hora de la venta.	FK → time_dim.t_time_sk
`ws_ship_date_sk`	`bigint`	Fecha de envío del pedido.	FK → date_dim.d_date_sk
`ws_item_sk`	`bigint`	Producto vendido; con `ws_order_number` forma la clave primaria.	PK (1/2); FK → item.i_item_sk
`ws_bill_customer_sk`	`bigint`	Cliente al que se factura el pedido.	FK → customer.c_customer_sk
`ws_bill_cdemo_sk`	`bigint`	Perfil demográfico del cliente al que se factura.	FK → customer_demographics.cd_demo_sk
`ws_bill_hdemo_sk`	`bigint`	Perfil del hogar del cliente al que se factura.	FK → household_demographics.hd_demo_sk
`ws_bill_addr_sk`	`bigint`	Dirección de facturación.	FK → customer_address.ca_address_sk
`ws_ship_customer_sk`	`bigint`	Cliente que recibe el envío.	FK → customer.c_customer_sk
`ws_ship_cdemo_sk`	`bigint`	Perfil demográfico del cliente que recibe el envío.	FK → customer_demographics.cd_demo_sk
`ws_ship_hdemo_sk`	`bigint`	Perfil del hogar del cliente que recibe el envío.	FK → household_demographics.hd_demo_sk
`ws_ship_addr_sk`	`bigint`	Dirección de envío.	FK → customer_address.ca_address_sk
`ws_web_page_sk`	`bigint`	Página web desde la que se hizo el pedido.	FK → web_page.wp_web_page_sk
`ws_web_site_sk`	`bigint`	Sitio web en el que se hizo el pedido.	FK → web_site.web_site_sk
`ws_ship_mode_sk`	`bigint`	Modo de envío del pedido.	FK → ship_mode.sm_ship_mode_sk
`ws_warehouse_sk`	`bigint`	Almacén desde el que se sirve el pedido.	FK → warehouse.w_warehouse_sk
`ws_promo_sk`	`bigint`	Promoción aplicada a la venta.	FK → promotion.p_promo_sk
`ws_order_number`	`bigint`	Número de pedido; con `ws_item_sk` forma la clave primaria.	PK (2/2)
`ws_quantity`	`integer`	Cantidad de unidades vendidas.	—
`ws_wholesale_cost`	`decimal(7,2)`	Coste mayorista por unidad.	—
`ws_list_price`	`decimal(7,2)`	Precio de catálogo por unidad.	—
`ws_sales_price`	`decimal(7,2)`	Precio de venta por unidad.	—
`ws_ext_discount_amt`	`decimal(7,2)`	Importe extendido del descuento aplicado.	—
`ws_ext_sales_price`	`decimal(7,2)`	Importe extendido de la venta antes de descuentos e impuestos.	—
`ws_ext_wholesale_cost`	`decimal(7,2)`	Coste mayorista extendido de los productos vendidos.	—
`ws_ext_list_price`	`decimal(7,2)`	Precio de lista extendido de los productos vendidos.	—
`ws_ext_tax`	`decimal(7,2)`	Impuesto extendido de la venta.	—
`ws_coupon_amt`	`decimal(7,2)`	Importe total de los cupones aplicados.	—
`ws_ext_ship_cost`	`decimal(7,2)`	Coste extendido del envío del pedido.	—
`ws_net_paid`	`decimal(7,2)`	Importe neto pagado por la venta.	—
`ws_net_paid_inc_tax`	`decimal(7,2)`	Importe neto pagado incluyendo impuestos.	—
`ws_net_paid_inc_ship`	`decimal(7,2)`	Importe neto pagado incluyendo el envío.	—
`ws_net_paid_inc_ship_tax`	`decimal(7,2)`	Importe neto pagado incluyendo envío e impuestos.	—
`ws_net_profit`	`decimal(7,2)`	Beneficio neto de la venta.	—

Columna	Tipo	Descripción	Clave / relación
`sr_returned_date_sk`	`bigint`	Fecha de la devolución.	FK → date_dim.d_date_sk
`sr_return_time_sk`	`bigint`	Hora de la devolución.	FK → time_dim.t_time_sk
`sr_item_sk`	`bigint`	Producto devuelto; con `sr_ticket_number` forma la clave primaria.	PK (1/2); FK → item.i_item_sk; FK → store_sales.ss_item_sk
`sr_customer_sk`	`bigint`	Cliente que hace la devolución.	FK → customer.c_customer_sk
`sr_cdemo_sk`	`bigint`	Perfil demográfico del cliente.	FK → customer_demographics.cd_demo_sk
`sr_hdemo_sk`	`bigint`	Perfil del hogar del cliente.	FK → household_demographics.hd_demo_sk
`sr_addr_sk`	`bigint`	Dirección del cliente.	FK → customer_address.ca_address_sk
`sr_store_sk`	`bigint`	Tienda en la que se hace la devolución.	FK → store.s_store_sk
`sr_reason_sk`	`bigint`	Motivo de la devolución.	FK → reason.r_reason_sk
`sr_ticket_number`	`bigint`	Número del ticket devuelto; con `sr_item_sk` forma la clave primaria.	PK (2/2); FK → store_sales.ss_ticket_number
`sr_return_quantity`	`integer`	Número de unidades devueltas.	—
`sr_return_amt`	`decimal(7,2)`	Importe de la devolución.	—
`sr_return_tax`	`decimal(7,2)`	Impuestos asociados a la devolución.	—
`sr_return_amt_inc_tax`	`decimal(7,2)`	Importe de la devolución incluyendo impuestos.	—
`sr_fee`	`decimal(7,2)`	Comisión cobrada por la devolución.	—
`sr_return_ship_cost`	`decimal(7,2)`	Coste del envío de la devolución.	—
`sr_refunded_cash`	`decimal(7,2)`	Importe devuelto en efectivo.	—
`sr_reversed_charge`	`decimal(7,2)`	Importe de cargos revertidos.	—
`sr_store_credit`	`decimal(7,2)`	Importe compensado mediante crédito de tienda.	—
`sr_net_loss`	`decimal(7,2)`	Pérdida neta causada por la devolución.	—

Columna	Tipo	Descripción	Clave / relación
`cr_returned_date_sk`	`bigint`	Fecha de la devolución.	FK → date_dim.d_date_sk
`cr_returned_time_sk`	`bigint`	Hora de la devolución.	FK → time_dim.t_time_sk
`cr_item_sk`	`bigint`	Producto devuelto; con `cr_order_number` forma la clave primaria.	PK (1/2); FK → item.i_item_sk; FK → catalog_sales.cs_item_sk
`cr_refunded_customer_sk`	`bigint`	Cliente al que se reembolsa.	FK → customer.c_customer_sk
`cr_refunded_cdemo_sk`	`bigint`	Perfil demográfico del cliente al que se reembolsa.	FK → customer_demographics.cd_demo_sk
`cr_refunded_hdemo_sk`	`bigint`	Perfil del hogar del cliente al que se reembolsa.	FK → household_demographics.hd_demo_sk
`cr_refunded_addr_sk`	`bigint`	Dirección del cliente al que se reembolsa.	FK → customer_address.ca_address_sk
`cr_returning_customer_sk`	`bigint`	Cliente que devuelve el producto.	FK → customer.c_customer_sk
`cr_returning_cdemo_sk`	`bigint`	Perfil demográfico del cliente que devuelve el producto.	FK → customer_demographics.cd_demo_sk
`cr_returning_hdemo_sk`	`bigint`	Perfil del hogar del cliente que devuelve el producto.	FK → household_demographics.hd_demo_sk
`cr_returning_addr_sk`	`bigint`	Dirección del cliente que devuelve el producto.	FK → customer_address.ca_address_sk
`cr_call_center_sk`	`bigint`	Centro de llamadas que tramitó la devolución.	FK → call_center.cc_call_center_sk
`cr_catalog_page_sk`	`bigint`	Página del catálogo del pedido devuelto.	FK → catalog_page.cp_catalog_page_sk
`cr_ship_mode_sk`	`bigint`	Modo de envío de la devolución.	FK → ship_mode.sm_ship_mode_sk
`cr_warehouse_sk`	`bigint`	Almacén que recibe la devolución.	FK → warehouse.w_warehouse_sk
`cr_reason_sk`	`bigint`	Motivo de la devolución.	FK → reason.r_reason_sk
`cr_order_number`	`bigint`	Número del pedido devuelto; con `cr_item_sk` forma la clave primaria.	PK (2/2); FK → catalog_sales.cs_order_number
`cr_return_quantity`	`integer`	Número de unidades devueltas.	—
`cr_return_amount`	`decimal(7,2)`	Importe de la devolución.	—
`cr_return_tax`	`decimal(7,2)`	Impuestos asociados a la devolución.	—
`cr_return_amt_inc_tax`	`decimal(7,2)`	Importe de la devolución incluyendo impuestos.	—
`cr_fee`	`decimal(7,2)`	Comisión cobrada por la devolución.	—
`cr_return_ship_cost`	`decimal(7,2)`	Coste del envío de la devolución.	—
`cr_refunded_cash`	`decimal(7,2)`	Importe devuelto en efectivo.	—
`cr_reversed_charge`	`decimal(7,2)`	Importe de cargos revertidos.	—
`cr_store_credit`	`decimal(7,2)`	Importe compensado mediante crédito de tienda.	—
`cr_net_loss`	`decimal(7,2)`	Pérdida neta causada por la devolución.	—

Columna	Tipo	Descripción	Clave / relación
`wr_returned_date_sk`	`bigint`	Fecha de la devolución.	FK → date_dim.d_date_sk
`wr_returned_time_sk`	`bigint`	Hora de la devolución.	FK → time_dim.t_time_sk
`wr_item_sk`	`bigint`	Producto devuelto; con `wr_order_number` forma la clave primaria.	PK (1/2); FK → item.i_item_sk; FK → web_sales.ws_item_sk
`wr_refunded_customer_sk`	`bigint`	Cliente al que se reembolsa.	FK → customer.c_customer_sk
`wr_refunded_cdemo_sk`	`bigint`	Perfil demográfico del cliente al que se reembolsa.	FK → customer_demographics.cd_demo_sk
`wr_refunded_hdemo_sk`	`bigint`	Perfil del hogar del cliente al que se reembolsa.	FK → household_demographics.hd_demo_sk
`wr_refunded_addr_sk`	`bigint`	Dirección del cliente al que se reembolsa.	FK → customer_address.ca_address_sk
`wr_returning_customer_sk`	`bigint`	Cliente que devuelve el producto.	FK → customer.c_customer_sk
`wr_returning_cdemo_sk`	`bigint`	Perfil demográfico del cliente que devuelve el producto.	FK → customer_demographics.cd_demo_sk
`wr_returning_hdemo_sk`	`bigint`	Perfil del hogar del cliente que devuelve el producto.	FK → household_demographics.hd_demo_sk
`wr_returning_addr_sk`	`bigint`	Dirección del cliente que devuelve el producto.	FK → customer_address.ca_address_sk
`wr_web_page_sk`	`bigint`	Página web desde la que se tramitó la devolución.	FK → web_page.wp_web_page_sk
`wr_reason_sk`	`bigint`	Motivo de la devolución.	FK → reason.r_reason_sk
`wr_order_number`	`bigint`	Número del pedido devuelto; con `wr_item_sk` forma la clave primaria.	PK (2/2); FK → web_sales.ws_order_number
`wr_return_quantity`	`integer`	Número de unidades devueltas.	—
`wr_return_amt`	`decimal(7,2)`	Importe de la devolución.	—
`wr_return_tax`	`decimal(7,2)`	Impuestos asociados a la devolución.	—
`wr_return_amt_inc_tax`	`decimal(7,2)`	Importe de la devolución incluyendo impuestos.	—
`wr_fee`	`decimal(7,2)`	Comisión cobrada por la devolución.	—
`wr_return_ship_cost`	`decimal(7,2)`	Coste del envío de la devolución.	—
`wr_refunded_cash`	`decimal(7,2)`	Importe devuelto en efectivo.	—
`wr_reversed_charge`	`decimal(7,2)`	Importe de cargos revertidos.	—
`wr_account_credit`	`decimal(7,2)`	Importe compensado mediante crédito en cuenta.	—
`wr_net_loss`	`decimal(7,2)`	Pérdida neta causada por la devolución.	—

Columna	Tipo	Descripción	Clave / relación
`inv_date_sk`	`bigint`	Fecha del recuento; con el producto y el almacén forma la clave primaria.	PK (1/3); FK → date_dim.d_date_sk
`inv_item_sk`	`bigint`	Producto inventariado; con la fecha y el almacén forma la clave primaria.	PK (2/3); FK → item.i_item_sk
`inv_warehouse_sk`	`bigint`	Almacén inventariado; con la fecha y el producto forma la clave primaria.	PK (3/3); FK → warehouse.w_warehouse_sk
`inv_quantity_on_hand`	`integer`	Unidades disponibles en inventario.	—