Reconstruyendo el star schema como tablas Iceberg

Descripción

Esta lección abre el catálogo único de este módulo y carga las tres tablas de Kiosko que no necesitan reproducir ningún mecanismo de evolución ni time travel: kiosko.fact_orders (la semana fija de siempre), kiosko.dim_store (con country poblada desde su primer commit, el resultado final del módulo 4) y kiosko.dim_date (una tabla completamente nueva, el calendario de agosto de 2026). Las tres se cargan en un solo append() cada una, sin ningún paso intermedio — el mecanismo de cómo se llega a ese estado final ya se enseñó, a fondo, en los módulos 1 y 4; esta lección solo lo aplica.

Conexión con el módulo. Esta lección responde la primera exigencia del brief de la lección 2: "un solo catálogo, con las tablas conviviendo". Abre kiosko_warehouse/ y kiosko_catalog.db una sola vez en este módulo — las lecciones 4, 5 y 6 van a seguir escribiendo sobre este mismo catálogo, sin volver a crear uno nuevo.

Una analogía: los tres cimientos que ya no cambian de forma

De las cinco tablas del lakehouse de Kiosko, tres son como los cimientos de un edificio ya terminado: una vez vertidos, no vuelven a cambiar de forma mientras el edificio esté en pie. fact_orders es el registro histórico de una semana que ya pasó — nunca va a tener una fila 41. dim_store tiene tres tiendas, con su país ya derivado — ninguna franquicia nueva entra en esta guía. dim_date es un calendario — agosto de 2026 no va a tener un día 32. Las otras dos tablas del lakehouse, dim_product y fact_orders_at_scale, sí tienen una historia que reconstruir — por eso viven en lecciones separadas (4 y 5). Esta lección vierte los tres cimientos, de una sola vez, sin ningún andamiaje adicional.

Ejemplo trabajado: fact_orders, dim_store y dim_date en un solo catálogo

Paso 1 — El catálogo único de este módulo

El mismo patrón de load_catalog() que ya usaste en cada módulo anterior — la diferencia esta vez es que este catálogo va a acumular cinco tablas, no una, a lo largo de las lecciones 3 a 6:

# kiosko_star_tables.py -- modulo 8, leccion 3
import os
from datetime import date, datetime, timedelta

import pyarrow as pa
import pyarrow.compute as pc
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import (
    BooleanType, DateType, DoubleType, IntegerType, NestedField, StringType, TimestampType,
)

from raw_orders import RAW_ORDERS

DIM_STORE_ROWS = [
    {"store_id": "S01", "store_name": "Kiosko Centro", "city": "Bogota", "country": "Colombia"},
    {"store_id": "S02", "store_name": "Kiosko Norte", "city": "Lima", "country": "Peru"},
    {"store_id": "S03", "store_name": "Kiosko Sur", "city": "Santiago", "country": "Chile"},
]
DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]

(raw_orders.py es exactamente el mismo archivo del módulo 1, lección 6 — las cuarenta órdenes fijas de Kiosko.)

Paso 2 — Los tres esquemas, declarados explícitos

kiosko.dim_store lleva country desde su primer NestedField — no hay ningún update_schema().add_column() en esta lección. Ese mecanismo —cómo se agrega una columna sin reescribir archivos existentes— ya lo demostró, paso a paso, el módulo 4; esta lección arranca directamente en el estado final:

FACT_ORDERS_SCHEMA = Schema(
    NestedField(field_id=1, name="order_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="store_id", field_type=StringType(), required=True),
    NestedField(field_id=3, name="product_id", field_type=StringType(), required=True),
    NestedField(field_id=4, name="quantity", field_type=IntegerType(), required=True),
    NestedField(field_id=5, name="unit_price", field_type=DoubleType(), required=True),
    NestedField(field_id=6, name="revenue", field_type=DoubleType(), required=True),
    NestedField(field_id=7, name="order_ts", field_type=TimestampType(), required=True),
)
DIM_STORE_SCHEMA = Schema(
    NestedField(field_id=1, name="store_id", field_type=StringType(), required=True),
    NestedField(field_id=2, name="store_name", field_type=StringType(), required=True),
    NestedField(field_id=3, name="city", field_type=StringType(), required=True),
    NestedField(field_id=4, name="country", field_type=StringType(), required=True),
)
DIM_DATE_SCHEMA = Schema(
    NestedField(field_id=1, name="date_key", field_type=IntegerType(), required=True),
    NestedField(field_id=2, name="calendar_date", field_type=DateType(), required=True),
    NestedField(field_id=3, name="day_of_week", field_type=StringType(), required=True),
    NestedField(field_id=4, name="month", field_type=IntegerType(), required=True),
    NestedField(field_id=5, name="quarter", field_type=IntegerType(), required=True),
    NestedField(field_id=6, name="year", field_type=IntegerType(), required=True),
    NestedField(field_id=7, name="is_weekend", field_type=BooleanType(), required=True),
)

El esquema de dim_date es, deliberadamente, idéntico en columnas al dim_date que data-modeling-for-analytics-guide (módulo 8) ya construyó sobre DuckDB —date_key como entero YYYYMMDD, calendar_date como fecha real, day_of_week/month/quarter/year/is_weekend derivados—. Iceberg agrega un tipo nuevo que DuckDB no necesitó declarar así de explícito: DateType(), el mismo tipo que vas a ver mapeado a pa.date32() del lado de PyArrow.

Paso 3 — Las funciones que construyen cada pyarrow.Table

def fact_orders_pa_table() -> pa.Table:
    schema = pa.schema([
        pa.field("order_id", pa.string(), nullable=False),
        pa.field("store_id", pa.string(), nullable=False),
        pa.field("product_id", pa.string(), nullable=False),
        pa.field("quantity", pa.int32(), nullable=False),
        pa.field("unit_price", pa.float64(), nullable=False),
        pa.field("revenue", pa.float64(), nullable=False),
        pa.field("order_ts", pa.timestamp("us"), nullable=False),
    ])
    rows = [
        {
            "order_id": order_id, "store_id": store_id, "product_id": product_id,
            "quantity": quantity, "unit_price": unit_price,
            "revenue": round(quantity * unit_price, 10),
            "order_ts": datetime.fromisoformat(ts),
        }
        for order_id, store_id, product_id, quantity, unit_price, ts in RAW_ORDERS
    ]
    return pa.Table.from_pylist(rows, schema=schema)


def dim_store_pa_table() -> pa.Table:
    schema = pa.schema([
        pa.field("store_id", pa.string(), nullable=False),
        pa.field("store_name", pa.string(), nullable=False),
        pa.field("city", pa.string(), nullable=False),
        pa.field("country", pa.string(), nullable=False),
    ])
    return pa.Table.from_pylist(DIM_STORE_ROWS, schema=schema)


def dim_date_pa_table(start_date: str, end_date: str) -> pa.Table:
    schema = pa.schema([
        pa.field("date_key", pa.int32(), nullable=False),
        pa.field("calendar_date", pa.date32(), nullable=False),
        pa.field("day_of_week", pa.string(), nullable=False),
        pa.field("month", pa.int32(), nullable=False),
        pa.field("quarter", pa.int32(), nullable=False),
        pa.field("year", pa.int32(), nullable=False),
        pa.field("is_weekend", pa.bool_(), nullable=False),
    ])
    start, end, rows = date.fromisoformat(start_date), date.fromisoformat(end_date), []
    current = start
    while current <= end:
        weekday_index = current.weekday()
        rows.append({
            "date_key": int(current.strftime("%Y%m%d")), "calendar_date": current,
            "day_of_week": DAY_NAMES[weekday_index], "month": current.month,
            "quarter": (current.month - 1) // 3 + 1, "year": current.year,
            "is_weekend": weekday_index >= 5,
        })
        current += timedelta(days=1)
    return pa.Table.from_pylist(rows, schema=schema)

dim_date_pa_table() genera todo agosto de 2026 —del 01 al 31—, no solo la semana real de las órdenes. Esto es intencional: un calendario dimensional real cubre el período completo del negocio, no solo las fechas donde hubo transacciones — exactamente la misma decisión que tomó data-modeling-for-analytics-guide cuando generó su propio dim_date.

Paso 4 — El catálogo, el namespace, y las tres cargas

def main() -> None:
    print("=== Kiosko: el star -- fact_orders, dim_store (con country), dim_date ===\n")

    warehouse_path = os.path.abspath("kiosko_warehouse")
    catalog_db_path = os.path.abspath("kiosko_catalog.db")
    os.makedirs(warehouse_path, exist_ok=True)
    catalog = load_catalog(
        "kiosko", type="sql",
        uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
    )
    catalog.create_namespace("kiosko")
    print(f"Paso 1/4 -- catalogo '{catalog.name}' y namespace 'kiosko' listos")

    fact_orders = catalog.create_table("kiosko.fact_orders", schema=FACT_ORDERS_SCHEMA)
    fact_orders.append(fact_orders_pa_table())
    fact_rows = fact_orders.scan().to_arrow()
    fact_revenue = round(sum(fact_rows.column("revenue").to_pylist()), 2)
    print(f"Paso 2/4 -- kiosko.fact_orders cargada: {fact_rows.num_rows} filas, revenue total {fact_revenue}")

    dim_store = catalog.create_table("kiosko.dim_store", schema=DIM_STORE_SCHEMA)
    dim_store.append(dim_store_pa_table())
    store_rows = sorted(dim_store.scan().to_arrow().to_pylist(), key=lambda r: r["store_id"])
    print(f"Paso 3/4 -- kiosko.dim_store cargada: {len(store_rows)} filas, con 'country' desde su primer commit "
          f"(sin pasar por la evolucion de esquema del modulo 4 -- esa evolucion ya quedo demostrada ahi)")
    for row in store_rows:
        print(f"           {row['store_id']}  {row['store_name']:<14} {row['city']:<10} country={row['country']}")

    dim_date = catalog.create_table("kiosko.dim_date", schema=DIM_DATE_SCHEMA)
    dim_date.append(dim_date_pa_table("2026-08-01", "2026-08-31"))
    date_rows = dim_date.scan().to_arrow()
    week_dates = pc.filter(
        date_rows, pc.and_(
            pc.greater_equal(date_rows.column("calendar_date"), pa.scalar(date(2026, 8, 3))),
            pc.less_equal(date_rows.column("calendar_date"), pa.scalar(date(2026, 8, 9))),
        ),
    )
    print(f"Paso 4/4 -- kiosko.dim_date cargada: {date_rows.num_rows} filas (agosto 2026 completo), "
          f"{week_dates.num_rows} de ellas cubren la semana real de Kiosko (03 al 09)")

    print("\n=== Verificacion final ===\n")
    assert fact_rows.num_rows == 40
    assert fact_revenue == 106.15
    assert len(store_rows) == 3
    assert {r["store_id"]: r["country"] for r in store_rows} == {
        "S01": "Colombia", "S02": "Peru", "S03": "Chile",
    }
    assert date_rows.num_rows == 31
    assert week_dates.num_rows == 7

    print("Todas las verificaciones pasaron:")
    print(f"  - kiosko.fact_orders: 40 filas, revenue={fact_revenue}")
    print("  - kiosko.dim_store: 3 filas, country poblado desde el primer commit (Colombia/Peru/Chile)")
    print("  - kiosko.dim_date: 31 filas (agosto 2026), 7 cubren la semana real de las ordenes")


if __name__ == "__main__":
    main()

Qué esperar (verificado corriendo python3 kiosko_star_tables.py real, en un directorio nuevo):

=== Kiosko: el star -- fact_orders, dim_store (con country), dim_date ===

Paso 1/4 -- catalogo 'kiosko' y namespace 'kiosko' listos
Paso 2/4 -- kiosko.fact_orders cargada: 40 filas, revenue total 106.15
Paso 3/4 -- kiosko.dim_store cargada: 3 filas, con 'country' desde su primer commit (sin pasar por la evolucion de esquema del modulo 4 -- esa evolucion ya quedo demostrada ahi)
           S01  Kiosko Centro  Bogota     country=Colombia
           S02  Kiosko Norte   Lima       country=Peru
           S03  Kiosko Sur     Santiago   country=Chile
Paso 4/4 -- kiosko.dim_date cargada: 31 filas (agosto 2026 completo), 7 de ellas cubren la semana real de Kiosko (03 al 09)

=== Verificacion final ===

Todas las verificaciones pasaron:
  - kiosko.fact_orders: 40 filas, revenue=106.15
  - kiosko.dim_store: 3 filas, country poblado desde el primer commit (Colombia/Peru/Chile)
  - kiosko.dim_date: 31 filas (agosto 2026), 7 cubren la semana real de las ordenes

Tres tablas, tres append(), tres snapshots — uno por tabla, ninguna evolución de esquema ni de partición en esta lección. El directorio kiosko_warehouse/kiosko/ queda con tres subcarpetas (fact_orders/, dim_store/, dim_date/), cada una con su propia cadena metadata → manifest list → manifest files → data files, exactamente como mapeó el módulo 2 — la diferencia es que, por primera vez en esta guía, las tres viven bajo el mismo kiosko_catalog.db.

Profundización: por qué dim_store no repite la evolución de esquema aquí

Alguien que solo vio el módulo 4 podría esperar que esta lección repita, paso a paso, add_column("country", ...) seguido de un overwrite() que la pueble. No lo hace, y la razón es la misma disciplina que ya viste en data-modeling-for-analytics-guide cuando su capstone reconstruyó dim_product_scd directamente con MERGE INTO, sin repetir cada corrida individual del módulo 4 de esa guía: un capstone reconstruye el estado final verificado, no el proceso completo que llevó hasta ahí. El módulo 4 de esta guía ya demostró, con assert propios, que add_column() no reescribe archivos de datos existentes, que country se puebla correctamente desde city, y que un snapshot anterior a la evolución sigue leyendo su propio esquema. Repetir esa demostración aquí no agregaría ninguna evidencia nueva — solo alargaría este módulo sin ganancia pedagógica. Lo que sí es nuevo en esta lección es algo que el módulo 4, en aislamiento, no podía mostrar: dim_store con country ya poblada, conviviendo en el mismo catálogo que fact_orders y dim_date, lista para un JOIN real.

Errores comunes

Crear un catálogo nuevo por cada lección de este módulo, como hicieron los módulos 1 a 7. Qué pasa: alguien, por costumbre, borra kiosko_warehouse/ y kiosko_catalog.db antes de correr la lección 4, esperando —como en cada módulo anterior— empezar desde cero. Por qué pasa: los siete módulos anteriores, cada uno, sí esperaban que reiniciaras el catálogo en cada proyecto de cierre; ese hábito es difícil de romper. Cómo detectarlo: si la lección 4 falla con TableDoesNotExistError al intentar catalog.load_table("kiosko.fact_orders"), borraste el catálogo que esta lección acaba de crear. Cómo corregirlo: a partir de esta lección, y hasta el final de este módulo (lecciones 3 a 6), no borres kiosko_warehouse/ ni kiosko_catalog.db entre lecciones — es la primera vez en esta guía en que varias lecciones consecutivas comparten, a propósito, el mismo catálogo.

Cargar dim_date solo con la semana real de órdenes (03 al 09 de agosto), en vez del mes completo. Qué pasa: alguien, para "ahorrar filas", genera dim_date solo para las fechas donde hay órdenes reales, en vez de todo agosto de 2026. Por qué pasa: parece más eficiente cargar solo lo que se va a usar en un JOIN. Cómo detectarlo: si tu kiosko.dim_date tiene 7 filas en vez de 31, y algún reporte futuro necesita un rango de fechas fuera de la semana de las órdenes (por ejemplo, "revenue por semana de todo agosto", con semanas vacías mostrando cero), tu calendario no va a poder responder esa pregunta. Cómo corregirlo: un dim_date dimensional cubre el período completo del negocio, no solo las fechas con actividad — es exactamente la misma decisión, con el mismo rango, que data-modeling-for-analytics-guide ya tomó.

Ejercicios

Ejercicio 1 — Corre el script tú mismo, desde cero. En un directorio nuevo, con raw_orders.py en el mismo lugar, corre python3 kiosko_star_tables.py. Confirma que ves los cuatro pasos completarse y el mensaje final con las tres verificaciones.

Ver solución

Si raw_orders.py está en el mismo directorio y PyIceberg está instalado, la salida debería reproducir exactamente la estructura de esta lección: cuatro pasos numerados, seguidos de la verificación final con 106.15 de revenue, los tres países correctos, y 31/7 para dim_date. No borres este directorio — la lección 4 continúa exactamente donde esta termina.

Ejercicio 2 — Rompe un assert a propósito, y observa el fallo. Cambia temporalmente el rango de dim_date_pa_table() de "2026-08-01"/"2026-08-31" a "2026-08-01"/"2026-08-15", corre el script de nuevo, y observa qué assert falla primero. Después revierte el cambio.

Ver solución

El primer assert en fallar es assert date_rows.num_rows == 31 — con el rango recortado a la primera quincena, dim_date tendría 15 filas, no 31. El assert week_dates.num_rows == 7 seguiría pasando, porque la semana real de las órdenes (03 al 09 de agosto) sigue completa dentro del rango recortado — este ejercicio confirma que los dos assert verifican cosas distintas: uno el tamaño total del calendario, el otro que la ventana de negocio relevante esté cubierta.

Ejercicio 3 — Explica, en tus propias palabras, por qué esta lección NO vuelve a ejecutar update_schema().add_column("country", ...). En 2-3 frases, justifica por qué reconstruir el estado final de dim_store es una decisión correcta para un capstone, y no un atajo que esconde información.

Ver solución

El módulo 4 de esta guía ya demostró, con assert propios sobre archivos de datos antes y después, que add_column("country", ...) no reescribe ningún Parquet existente — esa evidencia ya existe y no necesita repetirse. Un capstone que reconstruye el estado final, en vez de repetir cada paso intermedio de cada módulo, sigue el mismo patrón que ya usaron data-modeling-for-analytics-guide y dbt-analytics-engineering-guide: la meta de un módulo de cierre es demostrar que las piezas ya verificadas conviven correctamente, no volver a probar, una por una, cada garantía individual que ya tiene su propia evidencia en un módulo dedicado.

Resumen y siguiente paso

En esta lección abriste el catálogo único de este módulo y cargaste las tres tablas de Kiosko que no cambian de estado: kiosko.fact_orders (40 filas, 106.15), kiosko.dim_store (3 filas, country ya poblada) y kiosko.dim_date (31 filas, agosto 2026 completo) — la primera vez en esta guía que tres tablas conviven en el mismo kiosko_catalog.db.

Antes de avanzar deberías poder: nombrar las tres tablas que esta lección cargó y explicar por qué ninguna de las tres necesita reproducir un mecanismo de evolución; y explicar por qué esta lección, a diferencia de los proyectos de módulos anteriores, no borra el catálogo entre lecciones.

La lección 4 —la pieza central de todo este capstone— crea kiosko.dim_product sin ninguna columna de historia, reproduce el cambio real de P002, y une las cuatro tablas del star para recuperar el margen correcto (10.8) con time travel puro.

Recursos

• PyIceberg — documentación oficial (quickstart), el flujo de load_catalog(), create_namespace(), create_table() y append() que integra esta lección. py.iceberg.apache.org. En inglés.
• PyIceberg — referencia de API, tipos (DateType, BooleanType, IntegerType), la base del esquema de dim_date. py.iceberg.apache.org/api. En inglés.
• DISEÑO de data-modeling-for-analytics-guide — fuente del esquema exacto de dim_date (date_key/calendar_date/day_of_week/month/quarter/year/is_weekend) que esta lección reconstruye sobre Iceberg. src/guides/data-modeling-for-analytics-guide/DISENO.md. En español.
• Esta misma guía, módulo 4, lección 5 — fuente del mecanismo completo de evolución de esquema que pobló country por primera vez. ../module-04-schema-evolution-without-rewriting/es/05-adding-country-to-dim-store.md. En español.
• DISEÑO de esta guía — el mapa completo de los ocho módulos, incluida la lección 4 que sigue. src/guides/lakehouse-and-iceberg-guide/DISENO.md. En español.