Trabajar con el historial de tablas

Para las tablas de Apache Iceberg y Delta Lake, cada operación que modifica una tabla crea una nueva versión de la tabla. Use la información del historial para auditar las operaciones, revertir una tabla o consultar una tabla en un momento dado mediante el desplazamiento de tiempo.

Recuperación del historial de tablas

Ejecute el DESCRIBE HISTORY comando para recuperar información, incluidas las operaciones, el usuario y la marca de tiempo de cada escritura en una tabla. Las operaciones se devuelven en orden cronológico inverso.

Para las columnas que DESCRIBE HISTORY devuelven, los valores en la operationParameters columna y las métricas por operación en la operationMetrics columna, véase Esquema histórico de tablas y métricas de operaciones.

La retención del historial de tablas viene determinada por la configuración de tabla logRetentionDuration, que es de 30 días de manera predeterminada.

DESCRIBE HISTORY table_name       -- get the full history of the table

DESCRIBE HISTORY table_name LIMIT 1  -- get the last operation only

Para obtener más información sobre la sintaxis de Spark SQL, consulte DESCRIBE HISTORY.

Para obtener información sobre la sintaxis de Scala, Java y Python, consulte la documentación de La API de Delta Lake.

El Explorador de catálogos muestra el historial de tablas visualmente en la pestaña Historial .

Identificación del tipo de OPTIMIZE operación

La compactación automática, la agrupación en clústeres líquidos y la ordenación Z aparecen en el historial de tablas como OPTIMIZE operaciones. Para determinar cuál de ellos se ejecutó, inspeccione la columna operationParameters.

Para clasificar cada OPTIMIZE operación en el historial de una tabla, ejecute lo siguiente:

SELECT
  version,
  timestamp,
  CASE
    WHEN operationParameters.clusterBy IS NOT NULL AND operationParameters.clusterBy <> '[]' THEN 'Liquid clustering'
    WHEN operationParameters.zOrderBy IS NOT NULL AND operationParameters.zOrderBy <> '[]' THEN 'Z-ordering'
    WHEN operationParameters.auto = 'true' THEN 'Auto compaction'
    ELSE 'Manual OPTIMIZE'
  END AS optimize_type,
  operationParameters.auto AS is_auto_compaction,
  operationParameters.clusterBy AS cluster_by,
  operationParameters.zOrderBy AS z_order_by,
  operationMetrics.numRemovedFiles AS files_compacted,
  operationMetrics.numAddedFiles AS files_added,
  operationMetrics.numRemovedBytes AS bytes_removed,
  operationMetrics.numAddedBytes AS bytes_added
FROM (DESCRIBE HISTORY table_name)
WHERE operation = 'OPTIMIZE'
ORDER BY version DESC;

En las secciones siguientes se describe cada operationParameters valor con detalle. Para definiciones de las operationMetrics claves que selecciona la consulta anterior, véase Métricas de operación.

Compactación automática

La compactación automática establece el parámetro auto en true. Azure Databricks desencadena la compactación automática automáticamente después de una escritura. Cuando auto es false, un usuario o un trabajo programado ejecutó el OPTIMIZE comando.

Por ejemplo, una operación de compactación automática muestra lo siguiente:

operationParameters: {
  "auto": "true"
}

Para obtener más información sobre la compactación automática, consulte Auto compactación.

Agrupación en clústeres líquidos

La agrupación líquida en clústeres rellena el parámetro clusterBy con los nombres de las columnas de agrupación en clústeres. Una matriz vacía clusterBy ([]) indica solo la compactación de archivos.

Por ejemplo, una operación que agrupa los datos por las date columnas y region muestra lo siguiente:

operationParameters: {
  "clusterBy": "[\"date\",\"region\"]"
}

Para obtener más información sobre la agrupación en clústeres líquidos, consulte Uso de clústeres líquidos para tablas.

Ordenación Z

El orden Z rellena el zOrderBy parámetro con los nombres de columna de orden Z. Una matriz vacía zOrderBy ([]) indica que la operación no aplicó el orden Z.

Por ejemplo, una operación que aplicó el orden Z en la date columna muestra lo siguiente:

operationParameters: {
  "zOrderBy": "[\"date\"]"
}

Ámbito de la operación

El predicate parámetro indica si la operación se ejecutó en la tabla completa o solo parte de ella:

• Una matriz vacía predicate ([]) significa que la operación se ejecutó en toda la tabla.
• Una matriz rellenada predicate significa que un comando de destino OPTIMIZE table_name WHERE <partition_predicate> se ejecutó solo en las particiones que coinciden con el predicado.

Por ejemplo, una operación dirigida a las particiones que coinciden con year = 2024 muestra lo siguiente:

operationParameters: {
  "predicate": "[\"'year = 2024\"]"
}

Viaje en el tiempo

El viaje en el tiempo admite la consulta de versiones anteriores de tablas basadas en la marca de tiempo o en la versión de la tabla, según se registra en el registro de transacciones. Puede usar el viaje en el tiempo para aplicaciones como las siguientes:

• Volver a crear análisis, informes o salidas, como la salida de un modelo de Machine Learning. Esto puede ser útil para depurar o auditar, especialmente en sectores regulados.
• Escribir consultas temporales complejas.
• Corregir errores en los datos.
• Proporcionar aislamiento de instantáneas a un conjunto de consultas para tablas que cambian rápidamente.

Sintaxis de viaje en el tiempo

Para consultar una tabla con desplazamiento de tiempo, agregue una cláusula después de la especificación de nombre de tabla.

• El valor de timestamp_expression puede ser uno de los siguientes:
  • '2018-10-18T22:15:12.013Z', es decir, una cadena que se puede convertir en una marca de tiempo
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', es decir, una cadena de fecha.
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Cualquier otra expresión que sea una marca de tiempo o se pueda convertir en una
• version es un valor largo que se puede obtener de la salida de DESCRIBE HISTORY table_spec.

Ni timestamp_expression ni version pueden ser subconsultas.

Solo se aceptan cadenas de fecha o de hora. Por ejemplo, "2019-01-01" y "2019-01-01T00:00:00.000Z". Consulte el código siguiente para obtener una sintaxis de ejemplo:

SQL

SELECT * FROM people10m TIMESTAMP AS OF '2018-10-18T22:15:12.013Z';
SELECT * FROM people10m VERSION AS OF 123;

Python

df1 = spark.read.option("timestampAsOf", "2019-01-01").table("people10m")
df2 = spark.read.option("versionAsOf", 123).table("people10m")

También puede usar la sintaxis @ para especificar la marca de tiempo o la versión como parte del nombre de la tabla. La marca de tiempo debe estar en formato yyyyMMddHHmmssSSS. Puede especificar una versión con @v. Consulte el código siguiente para obtener una sintaxis de ejemplo:

SQL

-- Timestamp version
SELECT * FROM people10m@20190101000000000
-- Version number
SELECT * FROM people10m@v123

Python

# Timestamp version
spark.read.table("people10m@20190101000000000")
# Version number
spark.read.table("people10m@v123")

Configurar la retención de datos para consultas históricas

Para consultar una versión de tabla anterior, debe conservar tanto el registro como los archivos de datos de esa versión:

• Los archivos de datos se eliminan cuando se ejecuta VACUUM contra una tabla.
• Los archivos de registro se quitan automáticamente después de las versiones de la tabla de puntos de control.

Para aumentar el umbral de retención de datos para las tablas, debe configurar las siguientes propiedades de tabla, reemplazando <format> por delta o iceberg:

• <format>.logRetentionDuration = "interval <interval>": controla cuánto tiempo se conserva el historial de una tabla. El valor predeterminado es interval 30 days.
  • En Databricks Runtime 18.0 y versiones posteriores, logRetentionDuration debe ser mayor o igual que deletedFileRetentionDuration. En el caso de las tablas administradas por el catálogo de Unity, esto se aplica a Databricks Runtime 12.2 y versiones posteriores.
• <format>.deletedFileRetentionDuration = "interval <interval>": determina el umbral que VACUUM usa para quitar los archivos de datos a los que ya no se hace referencia en la versión actual de la tabla. El valor predeterminado es interval 7 days.

Por ejemplo, para acceder a 30 días de datos históricos, establezca delta.deletedFileRetentionDuration = "interval 30 days", que coincida con la configuración predeterminada de delta.logRetentionDuration.

Puede especificar propiedades de tabla durante la creación de tablas o establecerlas con una ALTER TABLE instrucción . Consulte Referencia de propiedades de tabla.

Ejemplos de viajes de tiempo

Para corregir eliminaciones accidentales en una tabla para el usuario 111:

INSERT INTO my_table
  SELECT * FROM my_table TIMESTAMP AS OF date_sub(current_date(), 1)
  WHERE userId = 111

Para corregir actualizaciones incorrectas accidentales en una tabla:

MERGE INTO my_table target
  USING my_table TIMESTAMP AS OF date_sub(current_date(), 1) source
  ON source.userId = target.userId
  WHEN MATCHED THEN UPDATE SET *

Para consultar el número de clientes nuevos agregados en la última semana:

SELECT
(
  SELECT count(distinct userId)
  FROM my_table
)
-
(
  SELECT count(distinct userId)
  FROM my_table TIMESTAMP AS OF date_sub(current_date(), 7)
) AS new_customers

Puntos de control del registro de transacciones

El registro de transacciones registra versiones de tabla como archivos JSON en el directorio del registro de transacciones junto con los datos de la tabla.

Para optimizar la consulta de puntos de comprobación, las versiones de tabla se agregan a los archivos de punto de comprobación de Parquet, lo que mejora el rendimiento evitando la necesidad de leer todas las versiones JSON del historial de tablas. Los usuarios no necesitan interactuar directamente con los puntos de control.

Azure Databricks optimiza la frecuencia de puntos de comprobación para el tamaño y la carga de trabajo de los datos. La frecuencia de los puntos de comprobación está sujeta a cambios sin previo aviso.

Restauración de una tabla a un estado anterior

Use el RESTORE comando para restaurar una tabla en una versión o marca de tiempo anterior, incluidos para estos escenarios:

• Puede restaurar una tabla ya restaurada.
• Puede restaurar una tabla clonada.

Tenga en cuenta los siguientes requisitos:

• Para restaurar una tabla, debe tener MODIFY permiso para la tabla.
• Una vez eliminados los archivos de datos, manualmente o por VACUUM, no se puede restaurar una tabla a una versión anterior que haga referencia a esos archivos. La restauración a esta versión parcialmente sigue siendo posible si spark.sql.files.ignoreMissingFiles se establece en true.
• Para restaurar por marca de tiempo, use los formatos yyyy-MM-dd HH:mm:ss o yyyy-MM-dd.

RESTORE TABLE target_table TO VERSION AS OF <version>;
RESTORE TABLE target_table TO TIMESTAMP AS OF <timestamp>;

Para obtener más información sobre la sintaxis, consulte RESTORE.

Comportamiento de streaming

La restauración es una operación que modifica los datos y podría provocar datos duplicados en las cargas de trabajo posteriores. Las entradas de registro agregadas por el RESTORE comando contienen dataChange establecido en true.

En el caso de las cargas de trabajo de bajada, como un trabajo de streaming estructurado que procesa las actualizaciones de una tabla, las entradas del registro de cambios de datos agregadas por la operación de restauración se consideran nuevas actualizaciones de datos y su procesamiento puede dar lugar a datos duplicados.

Por ejemplo:

En el ejemplo anterior, el RESTORE comando da como resultado actualizaciones que se vieron anteriormente al leer la versión 0 y 1 de la tabla. Si una consulta de streaming vuelve a leer esta tabla, estos archivos se consideran datos recién agregados y se vuelven a procesar.

Restaurar métricas

Una vez completado, RESTORE devuelve las siguientes métricas en un DataFrame de una sola fila:

• table_size_after_restore: tamaño de la tabla después de la restauración.

• num_of_files_after_restore: número de archivos de la tabla después de la restauración.

• num_removed_files: número de archivos quitados (eliminados lógicamente) de la tabla.

• num_restored_files: número de archivos restaurados debido a la reversión.

• removed_files_size: tamaño total en bytes de los archivos que se han quitado de la tabla.

• restored_files_size: tamaño total en bytes de los archivos que se han restaurado.

  

Buscar la última versión del commit

Para obtener el número de versión de la última confirmación escrita por el elemento SparkSession actual en todos los subprocesos y todas las tablas, consulte la configuración spark.databricks.<format>.lastCommitVersionInSession de SQL. Reemplace <format> por o deltaiceberg, según el formato de la tabla.

Por ejemplo:

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

Scala

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

Si SparkSession no ha realizado ninguna confirmación, consultar la clave devuelve un valor vacío.