Módulo 4: Schema Evolution Without Rewriting

Agregando una columna sin reescribir datos

Siguiente →

Descripción

Esta lección crea la tabla nueva de este módulo, kiosko.dim_store, con las tres tiendas de Kiosko y sus tres columnas originales —store_id, store_name, city—. Después le agrega una cuarta columna, country, con table.update_schema(). Vas a confirmar, con evidencia real —no con una afirmación de la documentación—, que esa operación no toca ni un solo byte de los archivos Parquet que ya existen, y que las tres filas ya cargadas quedan, por ahora, con country=None. Poblar esa columna con los tres países reales de Kiosko es el trabajo de la lección 5 — esta lección, a propósito, se detiene antes de ese paso, para aislar el mecanismo puro de la evolución de esquema.

Conexión con el módulo. La lección 2 demostró que una escritura Iceberg —un cambio de datos— es atómica. Esta lección introduce el otro tipo de cambio que Iceberg soporta: un cambio de esquema, que ni siquiera necesita tocar los archivos de datos para completarse. Es la primera vez en esta guía que table.update_schema() aparece — el módulo 3 ya lo había nombrado, en su profundización sobre "qué cuenta como una escritura", pero nunca lo había ejecutado.

Una analogía: agregar la pregunta al formulario, sin volver a tocar ninguna puerta

Retomando la analogía de la lección 1: kiosko.dim_store es, en este punto, un censo ya completo de tres casas —S01, S02, S03—, con un formulario de tres preguntas. Esta lección agrega una cuarta pregunta al formulario —country—, pero no envía a nadie a tocar de nuevo ninguna de las tres puertas. Las tres fichas ya archivadas quedan, por ahora, con la casilla de la pregunta nueva en blanco — no porque el censo esté roto, sino porque nadie fue todavía a completarla. El formulario cambió; las fichas ya archivadas, no.

Ejemplo trabajado: dim_store, y su primera evolución de esquema

Paso 1 — Crea kiosko.dim_store, con tres columnas

Con el mismo catálogo kiosko de las tres guías anteriores —un SqlCatalog respaldado por SQLite, warehouse en el filesystem local—, declara el esquema original:

# create_dim_store.py
import os

from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import NestedField, StringType

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

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),
)

table = catalog.create_table("kiosko.dim_store", schema=dim_store_schema)
print("Tabla creada:", table.name())
print()
print(table.schema())

Qué esperar (verificado corriendo el script real):

Tabla creada: ('kiosko', 'dim_store')

table {
  1: store_id: required string
  2: store_name: required string
  3: city: required string
}

Paso 2 — Carga las tres tiendas, y captura snap_before_evolution

# load_stores.py -- las tres tiendas de Kiosko, sin ninguna evolucion todavia
import os

import pyarrow as pa
from pyiceberg.catalog import load_catalog

warehouse_path = os.path.abspath("kiosko_warehouse")
catalog_db_path = os.path.abspath("kiosko_catalog.db")

catalog = load_catalog(
    "kiosko", type="sql",
    uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
)
table = catalog.load_table("kiosko.dim_store")

dim_store_pa_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),
])

DIM_STORE = [
    {"store_id": "S01", "store_name": "Kiosko Centro", "city": "Bogota"},
    {"store_id": "S02", "store_name": "Kiosko Norte", "city": "Lima"},
    {"store_id": "S03", "store_name": "Kiosko Sur", "city": "Santiago"},
]

pa_table = pa.Table.from_pylist(DIM_STORE, schema=dim_store_pa_schema)
table.append(pa_table)

# capturado DE INMEDIATO, siguiendo la regla del modulo 3 -- nunca hardcodear un snapshot-id
snap_before_evolution = table.current_snapshot().snapshot_id

print("table.scan().to_arrow() tras la carga:")
for row in table.scan().to_arrow().to_pylist():
    print(f"  {row['store_id']}  {row['store_name']:<15} {row['city']}")
print()
print("snap_before_evolution capturado, type:", type(snap_before_evolution).__name__)
print("table.history() tiene", len(table.history()), "entrada(s)")

Qué esperar (verificado corriendo el script real; snap_before_evolution es un entero asignado por Iceberg en el momento del commit, distinto en cada corrida — nunca se hardcodea):

table.scan().to_arrow() tras la carga:
  S01  Kiosko Centro   Bogota
  S02  Kiosko Norte    Lima
  S03  Kiosko Sur      Santiago

snap_before_evolution capturado, type: int
table.history() tiene 1 entrada(s)

Guarda snap_before_evolution: es la foto de dim_store antes de cualquier evolución de esquema — la que la lección 6 de este módulo va a leer, para confirmar que sigue viéndose con su esquema original de tres columnas, incluso después de que la tabla cambie.

Paso 3 — table.update_schema().add_column("country", ...), y la prueba de que ningún archivo se toca

# add_country_column.py -- la evolucion de esquema en si
import os

from pyiceberg.catalog import load_catalog
from pyiceberg.types import StringType

warehouse_path = os.path.abspath("kiosko_warehouse")
catalog_db_path = os.path.abspath("kiosko_catalog.db")

catalog = load_catalog(
    "kiosko", type="sql",
    uri=f"sqlite:///{catalog_db_path}", warehouse=f"file://{warehouse_path}",
)
table = catalog.load_table("kiosko.dim_store")

files_before = sorted(f["file_path"].split("/")[-1] for f in table.inspect.files().to_pylist())
snapshots_before = len(table.history())

with table.update_schema() as update:
    update.add_column("country", StringType())

print("Esquema tras add_column('country', StringType()):")
print(table.schema())
print()

files_after = sorted(f["file_path"].split("/")[-1] for f in table.inspect.files().to_pylist())
snapshots_after = len(table.history())

print("Archivos de datos ANTES del add_column:", files_before)
print("Archivos de datos DESPUES del add_column:", files_after)
print("Mismos archivos (ningun Parquet reescrito):", files_before == files_after)
print()
print("table.history() ANTES:", snapshots_before, " DESPUES:", snapshots_after,
      "-- add_column NO crea un snapshot de datos")
print()
print("table.scan().to_arrow() -- filas existentes tras agregar la columna:")
for row in table.scan().to_arrow().to_pylist():
    print(f"  {row['store_id']}  {row['store_name']:<15} {row['city']:<10} country={row['country']}")

Qué esperar (verificado corriendo el script real):

Esquema tras add_column('country', StringType()):
table {
  1: store_id: required string
  2: store_name: required string
  3: city: required string
  4: country: optional string
}

Archivos de datos ANTES del add_column: ['00000-0-977f7238-b846-4c56-955e-12ac07baf5b5.parquet']
Archivos de datos DESPUES del add_column: ['00000-0-977f7238-b846-4c56-955e-12ac07baf5b5.parquet']
Mismos archivos (ningun Parquet reescrito): True

table.history() ANTES: 1  DESPUES: 1 -- add_column NO crea un snapshot de datos

table.scan().to_arrow() -- filas existentes tras agregar la columna:
  S01  Kiosko Centro   Bogota     country=None
  S02  Kiosko Norte    Lima       country=None
  S03  Kiosko Sur      Santiago   country=None

(El nombre exacto del archivo Parquet es el de tu propia corrida, distinto cada vez — lo que importa es que la lista antes y después es idéntica.)

Tres hechos, todos confirmados con evidencia, no con una promesa: 1) el esquema ahora tiene cuatro columnas, con country marcada optional (a diferencia de las otras tres, required) — vas a ver por qué esto es obligatorio en la Profundización de esta lección. 2) la lista de archivos de datos es exactamente la misma antes y después — el mismo archivo Parquet de tres filas, sin un byte reescrito. 3) table.history() no ganó ninguna entrada nueva — add_column no es una escritura de datos, así que no produce ningún snapshot, coherente con la distinción que el módulo 3 ya adelantó. Y las tres filas existentes, al leerlas después del cambio, muestran country=None — el "casillero en blanco" de la analogía de esta lección.

Diagrama: qué cambia, y qué no

flowchart LR
    A["kiosko.dim_store\n3 columnas, 3 filas\n1 archivo Parquet"] -->|"table.update_schema()\nadd_column('country')"| B["kiosko.dim_store\n4 columnas, 3 filas\nMISMO archivo Parquet"]
    A -.->|"snapshots"| S1["1 snapshot\n(el append)"]
    B -.->|"snapshots"| S1
    B -->|"country de las 3 filas"| N["None, None, None\n(pendiente de poblar -- leccion 5)"]

El único archivo que cambia es el de metadata —un JSON nuevo, con el esquema de cuatro columnas y una referencia al mismo snapshot y al mismo manifest de siempre—. El archivo de datos —el Parquet con las tres filas— nunca se toca.

Profundización: por qué la columna nueva tiene que ser optional

Fíjate en un detalle del esquema resultante: country quedó marcada optional, mientras que store_id, store_name y city son required. Esto no es una elección de estilo — es una consecuencia obligatoria de lo que esta lección acaba de demostrar. Si country fuera required sin un valor por defecto, las tres filas que ya existen tendrían que tener, de inmediato, un valor real ahí — pero esta lección probó que add_column no toca ningún archivo de datos, así que no hay ningún mecanismo para llenar ese valor en las filas viejas en el mismo instante en que se agrega la columna. La documentación oficial de Apache Iceberg, en su página "Evolution", lo resume así: "Iceberg schema updates are metadata changes, so no data files need to be rewritten to perform the update." Una columna required sin reescribir datos es, literalmente, una contradicción — por eso PyIceberg exige, por defecto, que cualquier columna agregada a una tabla con filas existentes sea optional (required=False es el valor por defecto de add_column()), salvo que le des un default_value explícito que Iceberg pueda usar para completar las filas viejas sin escribir nada.

La misma documentación oficial es explícita sobre la garantía de fondo, bajo el título "Correctness": "Added columns never read existing values from another column [...] Iceberg uses unique IDs to track each column in a table. When you add a column, it is assigned a new ID so existing data is never used by mistake." Esa es la razón técnica por la que country recibió el field_id=4 en el esquema de esta lección, nunca reutilizando el 1, 2 o 3 que ya tenían las columnas originales — el mismo mecanismo de field_id que el módulo 1 de esta guía ya presentó como la columna vertebral de cómo Iceberg identifica cada dato, ahora aplicado a por qué una columna nueva nunca puede, por accidente, heredar valores de una columna vieja.

Errores comunes

Llamar a update.add_column() fuera del bloque with table.update_schema() as update:. Qué pasa: alguien escribe update = table.update_schema(), después update.add_column(...), pero nunca hace commit, y el esquema de la tabla nunca cambia — sin ningún error visible que avise del problema. Por qué pasa: update_schema() devuelve un objeto UpdateSchema que acumula cambios, pero no los aplica hasta que algo confirma la transacción; sin el with, ese "algo" nunca ocurre. Cómo detectarlo: si después de llamar a add_column() vuelves a imprimir table.schema() y sigue mostrando las columnas viejas, tu cambio nunca se confirmó. Cómo corregirlo: usa siempre el patrón with table.update_schema() as update: update.add_column(...) — el bloque with es el que llama a commit() automáticamente al salir, exactamente el mismo patrón de administrador de contexto que ya usaste, sin saberlo, en cualquier with open(...) as f: de Python.

Esperar que country tenga los valores reales de país inmediatamente después de add_column. Qué pasa: alguien corre el paso 3 de esta lección, ve country=None en las tres filas, y concluye que algo salió mal. Por qué pasa: es natural esperar que "agregar una columna derivada de otra" complete el cálculo en el mismo paso. Cómo detectarlo: si tu objetivo es ver country='Colombia' para S01 inmediatamente después de este paso, revisa el mapa de la lección 1 — la población real de country es, a propósito, un paso separado. Cómo corregirlo: nada que corregir en esta lección — country=None para las tres filas existentes es el resultado correcto y esperado del paso 3. La lección 5 de este módulo hace la población real, con un overwrite() normal, exactamente como el módulo 3 pobló el cambio de P002.

Ejercicios

Ejercicio 1 — Reproduce los tres pasos tú mismo, y confirma la lista de archivos idéntica. En un directorio nuevo, corre los tres scripts de esta lección en orden. Confirma que files_before y files_after del paso 3 son exactamente la misma lista (un solo archivo Parquet), y que table.history() sigue en 1 después del add_column.

Ver solución

Si seguiste los tres pasos en el mismo directorio, tu salida debería coincidir en estructura con la de esta lección: el nombre exacto del archivo Parquet va a ser distinto (incluye un UUID generado en tu propia corrida), pero files_before == files_after debe dar True, y table.history() debe reportar 1 tanto antes como después del add_column. Si ves 2 después del add_column, revisa que no hayas llamado, por error, a table.append() o table.overwrite() en algún punto intermedio.

Ejercicio 2 — Predicción: ¿qué pasaría si intentaras agregar country como columna required, sin default_value? Sin correrlo todavía, busca en la referencia de API de PyIceberg (add_column(path, field_type, doc=None, required=False, default_value=None)) qué pasaría si llamaras update.add_column("country", StringType(), required=True) sobre una tabla que ya tiene tres filas cargadas, sin especificar default_value.

Ver solución

PyIceberg rechaza esa operación con un error explícito, porque agregar una columna required sin un valor por defecto violaría la garantía de la Profundización de esta lección: las tres filas existentes no tendrían ningún valor válido para una columna que el esquema dice que debe tener un valor siempre. La única forma de agregar una columna required sobre una tabla con datos ya cargados es proporcionar un default_value que Iceberg pueda aplicar, a nivel de metadata, a las filas existentes — sin necesidad de reescribirlas físicamente, pero de forma que cualquier lector sepa qué valor asumir para ellas.

Ejercicio 3 — Explica, con tus propias palabras, por qué el field_id de country es 4 y no 1, 2 o 3. En 2-3 frases, usando la cita de la documentación oficial de esta lección, explica por qué Iceberg nunca reutiliza un field_id ya usado, incluso si ese número correspondiera a una columna que ya no existe.

Ver solución

Iceberg identifica cada columna por su field_id, no por su nombre ni por su posición — es el mecanismo, ya presentado en el módulo 1 de esta guía, que hace posible renombrar o reordenar columnas sin ambigüedad. Si un field_id se reutilizara alguna vez, un archivo Parquet viejo que use ese número para referirse a una columna distinta podría, por accidente, "resucitar" con datos que nunca tuvieron ninguna relación con la columna nueva — exactamente la clase de error silencioso que la cita oficial describe: "so existing data is never used by mistake". Asignar siempre el siguiente número disponible, sin reciclar ninguno, es la forma más simple de garantizar que eso nunca pase.

Resumen y siguiente paso

En esta lección creaste kiosko.dim_store con sus tres columnas originales, capturaste snap_before_evolution antes de cualquier cambio, y agregaste la columna country con table.update_schema(). Confirmaste, con evidencia real —la misma lista de archivos Parquet antes y después, el mismo conteo de snapshots—, que esa operación es puramente de metadata: no reescribe, no toca, no crea ningún archivo de datos nuevo. Y viste por qué la columna nueva tiene que ser optional, y por qué recibe un field_id que nunca se reutilizó.

Antes de avanzar deberías poder: agregar una columna a una tabla Iceberg existente con table.update_schema(); explicar, con evidencia propia, por qué esa operación no reescribe archivos de datos; y explicar por qué una columna agregada sobre una tabla con filas existentes tiene que ser optional salvo que declares un default_value.

kiosko.dim_store ahora tiene cuatro columnas, con country todavía vacía. La lección 4 agrega dos operaciones más al mismo mecanismo —rename_column() y delete_column()— con una columna de práctica, temp_notes, agregada y borrada en la misma lección.

Recursos

  • Apache Iceberg — documentación oficial, "Evolution", sección "Schema evolution" y "Correctness", fuente de las dos citas de esta lección sobre metadata changes y unique field IDs. iceberg.apache.org/docs/latest/evolution. En inglés.
  • PyIceberg — referencia de API, table.update_schema() y UpdateSchema.add_column(path, field_type, doc=None, required=False, default_value=None). py.iceberg.apache.org/api. En inglés.
  • DISEÑO de data-engineering-foundations-guide — fuente del esquema exacto de stores (store_id, store_name, city) que esta lección carga sin cambios, antes de agregar country. src/guides/data-engineering-foundations-guide/DISENO.md. En español.
  • DISEÑO de esta guía — la sección "Evolución de esquema" (M4), fuente exacta de la secuencia dim_store → snap_before_evolution → add_column('country') que esta lección ejecuta. src/guides/lakehouse-and-iceberg-guide/DISENO.md. En español.
← AnteriorPor qué overwrite-partition nunca fue atómicoSiguiente →Renombrando y borrando columnas de forma segura