Actualización de esquemas de tabla con evolución del esquema

Las tablas admiten la evolución del esquema, lo que permite modificaciones en la estructura de tablas a medida que cambian los requisitos de datos. Se admiten los siguientes tipos de cambios:

Realice estos cambios explícitamente mediante DDL o implícitamente mediante DML.

Importante

Las actualizaciones de esquema entran en conflicto con todas las operaciones de escritura simultáneas. Databricks recomienda coordinar los cambios de esquema para evitar conflictos de escritura.

La actualización de un esquema de tabla finaliza las secuencias que se leen desde esa tabla. Para continuar con el procesamiento, reinicie la secuencia mediante los métodos descritos en Consideraciones de producción para Structured Streaming.

Cambios manuales de esquema

Use ALTER TABLE instrucciones para cambiar explícitamente el esquema de una tabla sin escribir nuevos datos.

Agregar columnas

Use ALTER TABLE ... ADD COLUMNS para agregar una o varias columnas a una tabla existente, especificando opcionalmente la posición y un comentario:

ALTER TABLE table_name ADD COLUMNS (col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

De manera predeterminada, la nulabilidad es true.

Ejemplo: Agregar campos anidados

La adición de columnas anidadas solo se admite para los structs. No se admiten matrices ni mapas.

Para agregar una columna a un campo anidado, use:

ALTER TABLE table_name ADD COLUMNS (col_name.nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

Por ejemplo, si el esquema antes de ejecutarse ALTER TABLE boxes ADD COLUMNS (colB.nested STRING AFTER field1) es:

- root
| - colA
| - colB
| +-field1
| +-field2

El esquema después es:

- root
| - colA
| - colB
| +-field1
| +-nested
| +-field2

Cambio de comentarios y ordenación de columnas

Use ALTER TABLE ... ALTER COLUMN para actualizar el comentario de una columna o reordenarla en relación con otras columnas:

ALTER TABLE table_name ALTER [COLUMN] col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Ejemplo: Cambiar campos anidados

Para cambiar una columna en un campo anidado, use:

ALTER TABLE table_name ALTER [COLUMN] col_name.nested_col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Por ejemplo, si el esquema antes de ejecutarse ALTER TABLE boxes ALTER COLUMN colB.field2 FIRST es:

- root
| - colA
| - colB
| +-field1
| +-field2

El esquema después es:

- root
| - colA
| - colB
| +-field2
| +-field1

Reemplazar columnas

Use ALTER TABLE ... REPLACE COLUMNS para volver a definir la lista de columnas completa de una tabla, incluida la adición, eliminación, reordenación o cambio de nombre de columnas en una sola operación:

ALTER TABLE table_name REPLACE COLUMNS (col_name1 col_type1 [COMMENT col_comment1], ...)

Ejemplo: Reemplazar campos anidados

Por ejemplo, al ejecutar el siguiente DDL:

ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT<field2:STRING, nested:STRING, field1:STRING>, colA STRING)

Si el esquema antes es:

- root
| - colA
| - colB
| +-field1
| +-field2

El esquema después es:

- root
| - colC
| - colB
| +-field2
| +-nested
| +-field1
| - colA

Cambiar el nombre de las columnas

Para cambiar el nombre de las columnas sin volver a escribir ninguno de los datos existentes de las columnas, debe habilitar la asignación de columnas para la tabla. Consulte Renombrar y eliminar columnas con el mapeo de columnas de Delta Lake.

Para cambiar el nombre de una columna:

ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name

Ejemplo: Cambiar el nombre de los campos anidados

Para cambiar el nombre de un campo anidado:

ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field

Por ejemplo, cuando ejecute el siguiente comando:

ALTER TABLE boxes RENAME COLUMN colB.field1 TO field001

Si el esquema antes es:

- root
| - colA
| - colB
| +-field1
| +-field2

El esquema después es:

- root
| - colA
| - colB
| +-field001
| +-field2

Consulte Renombrar y eliminar columnas con el mapeo de columnas de Delta Lake.

Eliminar columnas

Para eliminar columnas como una operación de solo metadatos sin volver a escribir ningún archivo de datos, debe habilitar la asignación de columnas para la tabla. Consulte Renombrar y eliminar columnas con el mapeo de columnas de Delta Lake.

Nota:

La eliminación de una columna de los metadatos no elimina los datos subyacentes de la columna en los archivos. Para purgar los datos de columna quitados:

  • Use REORG TABLE para reescribir archivos.
  • Después, use VACUUM para eliminar físicamente los archivos que contienen los datos de columna quitados.

Para eliminar una columna:

ALTER TABLE table_name DROP COLUMN col_name

Para eliminar varias columnas:

ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)

Cambiar el tipo de columna o el nombre

Puede cambiar el tipo o el nombre de una columna o quitar una columna reescribiendo la tabla. Para ello, use la opción overwriteSchema.

En el ejemplo siguiente se muestra cómo cambiar un tipo de columna:

(spark.read.table(...)
  .withColumn("birthDate", col("birthDate").cast("date"))
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

En el ejemplo siguiente se muestra cómo cambiar un nombre de columna:

(spark.read.table(...)
  .withColumnRenamed("dateOfBirth", "birthDate")
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

Habilitación de la evolución del esquema

Utilice WITH SCHEMA EVOLUTION o establezca mergeSchema en true para realizar cambios en el esquema en función del esquema de los datos que desee INSERT o MERGE en una tabla existente.

Habilite la evolución del esquema mediante uno de los métodos siguientes:

Databricks recomienda habilitar la evolución del esquema para cada operación de escritura mediante la WITH SCHEMA EVOLUTION sintaxis o la mergeSchema opción en lugar de establecer una configuración de Spark.

Cuando se usan opciones o sintaxis para habilitar la evolución del esquema en una operación de escritura, esto tiene prioridad sobre la configuración de Spark.

Habilitar la evolución del esquema al escribir para agregar nuevas columnas

Cuando se habilita la evolución del esquema, las columnas presentes en la consulta de origen, pero que faltan en la tabla de destino se agregan automáticamente como parte de una transacción de escritura. Consulte Habilitación de la evolución del esquema.

Ten en cuenta lo siguiente:

  • Las mayúsculas y minúsculas se conservan al anexar una nueva columna.
  • Las nuevas columnas se añaden al final del esquema de la tabla.
  • Si las columnas adicionales están en una estructura, se anexan al final de la estructura de la tabla de destino.

INSERT con la evolución del esquema mediante SQL

En Databricks Runtime 18.1 y versiones posteriores, use la cláusula WITH SCHEMA EVOLUTION en las instrucciones INSERT para habilitar la evolución del esquema:

INSERT WITH SCHEMA EVOLUTION INTO target_table
SELECT * FROM source_table

Si la consulta de source_table devuelve columnas que no existen en la tabla de destino, esas columnas se agregan automáticamente al target_table esquema. Las filas existentes reciben NULL valores para las nuevas columnas.

La cláusula WITH SCHEMA EVOLUTION admite las formas INSERT INTO, INSERT OVERWRITE y INSERT INTO ... REPLACE. El destino debe ser una tabla de Delta Lake o de Apache Iceberg. La inserción en una tabla de Hive u otra tabla no delta con esta cláusula devuelve un error.

Si usa Databricks Runtime 18.0 o una versión anterior, habilite la evolución del esquema con la opción mergeSchema en su lugar. Consulte INSERTsobre la evolución de esquemas con la API de DataFrame.

INSERT con la evolución del esquema mediante DataFrame API

En el ejemplo siguiente se muestra cómo usar la opción mergeSchema con una operación de escritura por lotes:

Python
(spark.read
  .table("source_table")
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("target_table")
)
Scala
spark.read
  .table("source_table")
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("target_table")

INSERT con la evolución del esquema con Structured Streaming

En el ejemplo siguiente se muestra el uso de la mergeSchema opción con Auto Loader para Structured Streaming. Consulte ¿Qué es Auto Loader?.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .trigger(availableNow=True)
  .toTable("table_name")
)

Evolución automática del esquema para la combinación

Para MERGE, la evolución del esquema permite resolver incompatibilidades de esquema entre las tablas de origen y de destino. Controla los dos casos siguientes:

  1. Existe una columna en la tabla de origen, pero no en la tabla de destino, y se especifica por nombre en una asignación de acciones de inserción o actualización. Como alternativa, una acción UPDATE SET * o INSERT * está presente.

    Esa columna se agregará al esquema de destino y sus valores se rellenarán desde la columna correspondiente del origen.

    • Esto solo se aplica cuando el nombre de columna y la estructura del origen de combinación coinciden exactamente con la asignación de destino.

    • La nueva columna debe estar presente en el esquema de origen. La asignación de la nueva columna en la cláusula de acción no define esa columna.

    Estos ejemplos permiten la evolución del esquema:

    -- The column newcol is present in the source but not in the target. It will be added to the target.
UPDATE SET target.newcol = source.newcol

-- The field newfield doesn't exist in struct column somestruct of the target. It will be added to that struct column.
UPDATE SET target.somestruct.newfield = source.somestruct.newfield

-- The column newcol is present in the source but not in the target.
-- It will be added to the target.
UPDATE SET target.newcol = source.newcol + 1

-- Any columns and nested fields in the source that don't exist in target will be added to the target.
UPDATE SET *
INSERT *

    Estos ejemplos no desencadenan la evolución del esquema si la columna newcol no está presente en el source esquema:

    UPDATE SET target.newcol = source.someothercol
UPDATE SET target.newcol = source.x + source.y
UPDATE SET target.newcol = source.output.newcol

  2. Existe una columna en la tabla de destino, pero no en la tabla de origen.

    No se cambia el esquema de destino. Estas columnas:

    • Se dejan sin cambios para UPDATE SET *.

    • Se establecen en NULL para INSERT *.

    • Aun así, puede modificarse explícitamente si se asigna en la cláusula de acción.

    Por ejemplo:

    UPDATE SET *  -- The target columns that are not in the source are left unchanged.
INSERT *  -- The target columns that are not in the source are set to NULL.
UPDATE SET target.onlyintarget = 5  -- The target column is explicitly updated.
UPDATE SET target.onlyintarget = source.someothercol  -- The target column is explicitly updated from some other source column.

Debe habilitar manualmente la evolución automática del esquema. Consulte Habilitación de la evolución del esquema.

Nota:

En Databricks Runtime 11.3 LTS y versiones posteriores, solo se pueden usar acciones INSERT * o UPDATE SET * para la evolución del esquema con combinación.

En Databricks Runtime 12.2 LTS y versiones posteriores, los campos de estructura y columnas presentes en la tabla de origen se pueden especificar por nombre en las acciones de inserción o actualización.

En Databricks Runtime 13.3 LTS y versiones posteriores, puede usar la evolución del esquema con estructuras anidadas dentro de mapas, como map<int, struct<a: int, b: int>>.

MERGEcon la evolución del esquema mediante SQL, Python y Scala

En Databricks Runtime 15.4 LTS y versiones posteriores, puede especificar la evolución del esquema en una instrucción merge mediante SQL o las API de tabla:

SQL
MERGE WITH SCHEMA EVOLUTION INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE
Python
from delta.tables import *

(targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)
Scala
import io.delta.tables._

targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

Operaciones de ejemplo de MERGE con evolución del esquema

Estos son algunos ejemplos de los efectos de la operación MERGE con y sin evolución del esquema.

Columns Consulta (en SQL) Comportamiento sin evolución del esquema (valor predeterminado) Comportamiento con evolución del esquema
Columnas de destino: key, value
Columnas de origen: key, value, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 El esquema de tabla permanece sin cambios; solo se actualizan o insertan las columnas key y value. El esquema de tabla se cambia a (key, value, new_value). Los registros existentes con coincidencias se actualizan con value y new_value en el origen. Las filas nuevas se insertan con el esquema (key, value, new_value).
Columnas de destino: key, old_value
Columnas de origen: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 Las acciones UPDATE y INSERT inician un error porque la columna de destino old_value no está en el origen. El esquema de tabla se cambia a (key, old_value, new_value). Los registros existentes con coincidencias se actualizan utilizando new_value del origen y se deja old_value sin modificar. Los nuevos registros se insertan con los valores key, new_value y NULL especificados para old_value.
Columnas de destino: key, old_value
Columnas de origen: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET new_value = s.new_value		 UPDATE genera un error porque la columna new_value no existe en la tabla de destino. El esquema de tabla se cambia a (key, old_value, new_value). Los registros existentes con coincidencias se actualizan con new_value en el origen y se deja old_value sin cambios. En los registros no coincidentes se especifica NULL para new_value. Consulte la nota (1).
Columnas de destino: key, old_value
Columnas de origen: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN NOT MATCHED
THEN INSERT (key, new_value) VALUES (s.key, s.new_value)		 INSERT genera un error porque la columna new_value no existe en la tabla de destino. El esquema de tabla se cambia a (key, old_value, new_value). Los nuevos registros se insertan con los valores key, new_value y NULL especificados para old_value. En los registros existentes se especifica NULL para new_value y se deja old_value sin cambios. Consulte la nota (1).

(1) Este comportamiento está disponible en Databricks Runtime 12.2 LTS y versiones posteriores; Databricks Runtime 11.3 LTS y versiones anteriores generan un error en esta situación.

Excluir columnas que participan en una combinación

En Databricks Runtime 12.2 LTS y versiones posteriores, puede usar cláusulas EXCEPT en condiciones de combinación para excluir explícitamente columnas. El comportamiento de la palabra clave EXCEPT varía en función de si está habilitada o no la evolución del esquema.

Cuando la evolución del esquema está desactivada, la palabra clave EXCEPT se aplica a la lista de columnas de la tabla de destino y permite excluir columnas de las acciones UPDATE o INSERT. Las columnas excluidas se establecen en null.

Con la evolución del esquema habilitada, la palabra clave EXCEPT se aplica a la lista de columnas de la tabla de origen y permite excluir columnas de la evolución del esquema. Una nueva columna en el origen, que no está presente en la tabla de destino, no se añade al esquema de destino si figura en la cláusula EXCEPT. Las columnas excluidas que ya están en el destino se configuran como null.

Ejemplos de EXCLUDE con MERGE

En los ejemplos siguientes se muestra esta sintaxis:

Columns Consulta (en SQL) Comportamiento sin evolución del esquema (valor predeterminado) Comportamiento con evolución del esquema
Columnas de destino: id, title, last_updated
Columnas de origen: id, title, review, last_updated		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated)		 Las filas coincidentes se actualizan estableciendo el campo last_updated en la fecha actual. Las filas nuevas se insertan mediante valores de id y title. El campo last_updated excluido se establece en null. El campo review se omite porque no está en el destino. Las filas coincidentes se actualizan estableciendo el campo last_updated en la fecha actual. El esquema ha evolucionado para agregar el campo review. Las filas nuevas se insertan con todos los campos de origen, excepto last_updated que se establece en null.
Columnas de destino: id, title, last_updated
Columnas de origen: id, title, review, internal_count		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated, internal_count)		 INSERT genera un error porque la columna internal_count no existe en la tabla de destino. Las filas coincidentes se actualizan estableciendo el campo last_updated en la fecha actual. El campo review se agrega a la tabla de destino, pero se omite el campo internal_count. Las filas nuevas insertadas tienen last_updated establecido en null.

Activar la evolución del esquema con la configuración de Spark (heredada)

Puede establecer la configuración spark.databricks.delta.schema.autoMerge.enabled en true para habilitar la evolución del esquema para todas las operaciones de escritura en la SparkSession actual.

Python

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", True)

Scala

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", true)

SQL

SET spark.databricks.delta.schema.autoMerge.enabled=true

Nota:

Databricks no recomienda este enfoque para producción. Establecer una configuración en toda la sesión podría provocar cambios de esquema no deseados en varias operaciones y dificulta la razón sobre qué operaciones evolucionan el esquema.

En su lugar, habilite la evolución del esquema para cada operación de escritura:

Cuando se usan opciones o sintaxis para habilitar la evolución del esquema en una operación de escritura, esto tiene prioridad sobre la configuración de Spark.

Reemplazar esquema de tabla

De forma predeterminada, sobrescribir los datos de una tabla no sobrescribe el esquema. Al sobrescribir una tabla con mode("overwrite") sin replaceWhere, es posible que desee sobrescribir también el esquema de los datos que se van a escribir.

Para reemplazar el esquema y la creación de particiones de la tabla, establezca la overwriteSchema opción en true:

df.write.option("overwriteSchema", "true")

Nota:

No se puede especificar overwriteSchema como true cuando se usa la sobrescritura de partición dinámica. Consulte Reescritura de particiones dinámicas con partitionOverwriteMode (heredado).

  • Last updated on