Trabalhar com o histórico da tabela

Cada operação que modifica uma tabela cria uma nova versão da tabela. Você pode usar as informações do histórico para auditar operações, reverter uma tabela ou consultar uma tabela em um ponto específico no tempo usando a viagem do tempo.

nota

O Databricks não recomenda o uso do histórico de tabelas como uma solução de backup de longo prazo para arquivamento de dados. Use apenas os últimos sete dias para operações de viagem do tempo, a menos que você tenha definido as configurações de retenção de dados e de log para um valor maior.

Recuperar história de uma tabela

Recupere informações, incluindo as operações, o usuário e o carimbo de data/hora de cada gravação em uma tabela executando o comando history. As operações são retornadas em ordem cronológica inversa.

A retenção do histórico da tabela é determinada pela configuração da tabela logRetentionDuration, que é de 30 dias por padrão.

nota

A viagem do tempo e o histórico da tabela são controlados por diferentes limites de retenção. Consulte O que é viagem do tempo?.

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

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

Para obter detalhes de sintaxe do Spark SQL, consulte DESCRIBE HISTORY.

Consulte a documentação da API do Delta Lake para obter detalhes da sintaxe Scala/Java/Python.

O Catalog Explorer fornece uma view destas informações detalhadas da tabela e o histórico. Além do esquema da tabela e dos dados de amostra, é possível clicar na tab Histórico para ver o histórico da tabela que é exibido com DESCRIBE HISTORY.

Esquema de história

A saída da operação history tem as seguintes colunas.

Coluna	Tipo	Descrição
version	long	Versão da tabela gerada pela operação.
carimbo de data/hora	carimbo de data/hora	Quando essa versão foi confirmada.
userId	string	ID do usuário que executou a operação.
userName	string	Nome do usuário que executou a operação.
operation	string	Nome da operação.
operationParameters	map	Parâmetros da operação (por exemplo, predicados.)
Job	struct	Detalhes do LakeFlow Job que executou a operação. É populado apenas para commits gerados por um LakeFlow Job. Caso contrário, `null`.
notebook	struct	Detalhes do notebook Databricks a partir do qual a operação foi executada. É preenchido apenas para commits escritos de um Notebook Databricks. Caso contrário, `null`.
clusterId	string	ID do cluster no qual a operação foi executada.
readVersion	long	Versão da tabela que foi lida para realizar a operação de gravação.
isolationLevel	string	Nível de isolamento usado para essa operação.
isBlindAppend	boolean	Se essa operação anexou dados.
operationMetrics	map	Métricas da operação (por exemplo, número de linhas e arquivos modificados).
userMetadata	string	Metadados de confirmação definidos pelo usuário, se tiverem sido especificados

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|   isolationLevel|isBlindAppend|    operationMetrics|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|      5|2019-07-29 14:07:47|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          4|WriteSerializable|        false|[numTotalRows -> ...|
|      4|2019-07-29 14:07:41|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          3|WriteSerializable|        false|[numTotalRows -> ...|
|      3|2019-07-29 14:07:29|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          2|WriteSerializable|        false|[numTotalRows -> ...|
|      2|2019-07-29 14:06:56|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          1|WriteSerializable|        false|[numTotalRows -> ...|
|      1|2019-07-29 14:04:31|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          0|WriteSerializable|        false|[numTotalRows -> ...|
|      0|2019-07-29 14:01:40|   ###|     ###|    WRITE|[mode -> ErrorIfE...|null|     ###|      ###|       null|WriteSerializable|         true|[numFiles -> 2, n...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

nota

Algumas das outras colunas não estarão disponíveis se você gravar em uma tabela usando os seguintes métodos:
As colunas adicionadas no futuro serão sempre adicionadas após a última coluna.

Compreendendo `partitionBy` em parâmetros de operação

O campo partitionBy é significativo apenas para operações de CREATE e OVERWRITE que definem ou alteram o esquema de partição de uma tabela.

Para operações de acréscimo em tabelas existentes (APPEND, INSERT, UPDATE, DELETE, MERGE), este campo pode exibir um array [] vazio ou colunas de partição, dependendo do método de escrita utilizado (.save() vs. .saveAsTable()). Esta inconsistência é o comportamento esperado e não deve ser usada para validar gravações.

importante

Não confie em partitionBy na história para validar operações de acréscimo. O valor varia com base nos detalhes de implementação, mas não afeta como os dados são gravados nas partições.

Exemplo

Considere uma tabela particionada pela coluna date:

Python
# Initial table creation - partitionBy is populated
df.write.format("delta") \
  .partitionBy("date") \
  .saveAsTable("sales_data")

A operações CREATE no história mostra:

operationParameters: {
  "mode": "ErrorIfExists",
  "partitionBy": "[\"date\"]"
}

Ao anexar dados a esta tabela:

Python
# Subsequent append - partitionBy shows empty
new_df.write.format("delta") \
  .mode("append") \
  .saveAsTable("sales_data")

A operação ANEXAR mostra:

operationParameters: {
  "mode": "Append",
  "partitionBy": "[]"
}

O valor partitionBy vazio é esperado. Os dados ainda são gravados nas partições corretas com base no esquema de partição existente da tabela. Observe que .save() para um caminho pode mostrar colunas de partição neste campo, mas essa diferença é um detalhe de implementação e não afeta o comportamento de escrita.

Métricas de operações

A operação history retorna uma coleção de métricas de operações no mapa de colunas operationMetrics .

As tabelas a seguir listam as principais definições do mapa por operação.

Operação	Nome da métrica	Descrição
WRITE, CREATE TABLE AS SELECT, REPLACE TABLE AS SELECT, COPY INTO
	numFiles	Número de arquivos gravados.
	numOutputBytes	Tamanho em bytes do conteúdo gravado.
	numOutputRows	Número de linhas gravadas.
STREAMING UPDATE
	numAddedFiles	Número de arquivos adicionados.
	numRemovedFiles	Número de arquivos removidos.
	numOutputRows	Número de linhas gravadas.
	numOutputBytes	Tamanho da gravação em bytes.
DELETE
	numAddedFiles	Número de arquivos adicionados. Não fornecido quando as partições da tabela são excluídas.
	numRemovedFiles	Número de arquivos removidos.
	numDeletedRows	Número de linhas removidas. Não fornecido quando as partições da tabela são excluídas.
	numCopiedRows	Número de linhas copiadas no processo de exclusão de arquivos.
	executionTimeMs	Tempo gasto para executar toda a operação.
	scanTimeMs	Tempo gasto para verificar os arquivos em busca de correspondências.
	rewriteTimeMs	Tempo gasto para regravar os arquivos correspondentes.
TRUNCATE
	numRemovedFiles	Número de arquivos removidos.
	executionTimeMs	Tempo gasto para executar toda a operação.
MERGE
	numSourceRows	Número de linhas no DataFrame de origem.
	numTargetRowsInserted	Número de linhas inseridas na tabela de destino.
	numTargetRowsUpdated	Número de linhas atualizadas na tabela de destino.
	numTargetRowsDeleted	Número de linhas excluídas na tabela de destino.
	numTargetRowsCopied	Número de linhas de destino copiadas.
	numOutputRows	Número total de linhas gravadas.
	numTargetFilesAdded	Número de arquivos adicionados ao coletor (destino).
	numTargetFilesRemoved	Número de arquivos removidos do coletor (destino).
	executionTimeMs	Tempo gasto para executar toda a operação.
	scanTimeMs	Tempo gasto para verificar os arquivos em busca de correspondências.
	rewriteTimeMs	Tempo gasto para regravar os arquivos correspondentes.
UPDATE
	numAddedFiles	Número de arquivos adicionados.
	numRemovedFiles	Número de arquivos removidos.
	numUpdatedRows	Número de linhas atualizadas.
	numCopiedRows	O número de linhas que acabaram de ser copiadas no processo de atualização de arquivos.
	executionTimeMs	Tempo gasto para executar toda a operação.
	scanTimeMs	Tempo gasto para verificar os arquivos em busca de correspondências.
	rewriteTimeMs	Tempo gasto para regravar os arquivos correspondentes.
FSCK	numRemovedFiles	Número de arquivos removidos.
CONVERT	numConvertedFiles	Número de arquivos Parquet que foram convertidos.
OPTIMIZE
	numAddedFiles	Número de arquivos adicionados.
	numRemovedFiles	Número de arquivos otimizados.
	numAddedBytes	Número de bytes adicionados depois que a tabela foi otimizada.
	numRemovedBytes	Número de bytes removidos.
	minFileSize	Tamanho do menor arquivo após a tabela ser otimizada.
	p25FileSize	Tamanho do arquivo do 25º percentil após a tabela ser otimizada.
	p50FileSize	Tamanho mediano do arquivo após a tabela ser otimizada.
	p75FileSize	Tamanho do arquivo do 75º percentil após a tabela ser otimizada.
	maxFileSize	Tamanho do maior arquivo após a tabela ser otimizada.
CLONE
	sourceTableSize	Tamanho em bytes da tabela de origem na versão clonada.
	sourceNumOfFiles	Número de arquivos na tabela de origem na versão clonada.
	numRemovedFiles	Número de arquivos removidos da tabela de destino se uma tabela anterior tiver sido substituída.
	removedFilesSize	Tamanho total em bytes dos arquivos removidos da tabela de destino se uma tabela anterior tiver sido substituída.
	numCopiedFiles	Número de arquivos que foram copiados para o novo local. 0 para clones rasos.
	copiedFilesSize	Tamanho total em bytes dos arquivos que foram copiados para o novo local. 0 para clones rasos.
RESTORE
	tableSizeAfterRestore	Tamanho da tabela em bytes após a restauração.
	numOfFilesAfterRestore	Número de arquivos na tabela após a restauração.
	numRemovedFiles	Número de arquivos removidos pela operação de restauração.
	numRestoredFiles	Número de arquivos adicionados como resultado da restauração.
	removedFilesSize	Tamanho em bytes dos arquivos removidos pela restauração.
	restoredFilesSize	Tamanho em bytes dos arquivos adicionados pela restauração.
VACUUM
	numDeletedFiles	Número de arquivos excluídos.
	numVacuumedDirectories	Número de diretórios aspirados.
	numFilesToDelete	Número de arquivos a serem excluídos.

O que é viagem do tempo?

A viagem do tempo é compatível com a consulta de versões anteriores da tabela com base no carimbo de data/hora ou na versão da tabela (conforme registrado no log de transações). Você pode usar a viagem do tempo para aplicações como as seguintes:

Recriar análises, relatórios ou resultados (por exemplo, o resultado de um modelo do machine learning).Isso pode ser útil para depuração ou auditoria, especialmente em indústrias regulamentadas.
Escrever consultas temporais complexas.
Corrigir erros em seus dados.
Fornecer isolamento de instantâneos para um conjunto de consultas para tabelas que mudam rapidamente.

importante

No Databricks Runtime 18.0 e acima, as consultas de viagem do tempo são bloqueadas se elas solicitarem uma versão mais antiga do que a propriedade de tabela deletedFileRetentionDuration (default 7 dias). Para tabelas gerenciadas pelo Unity Catalog, isso se aplica ao Databricks Runtime 12.2 e acima.

Sintaxe da viagem do tempo

Você consulta uma tabela com viagem do tempo adicionando uma cláusula após a especificação do nome da tabela.

timestamp_expression pode ser qualquer um dos seguintes:
- '2018-10-18T22:15:12.013Z', isto é, uma string que pode ser convertida em um carimbo de data/hora
- cast('2018-10-18 13:36:32 CEST' as timestamp)
- '2018-10-18', ou seja, uma string de data
- current_timestamp() - interval 12 hours
- date_sub(current_date(), 1)
- Qualquer outra expressão que seja ou possa ser convertida em um carimbo de data/hora
version é um valor longo que pode ser obtido da saída de DESCRIBE HISTORY table_spec.

Nem timestamp_expression nem version podem ser subconsultas.

Somente strings de data ou carimbo de data/hora são aceitas. Por exemplo, "2019-01-01" e "2019-01-01T00:00:00.000Z". Consulte o seguinte código para ver um exemplo de sintaxe:

SQL
Python

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

Você também pode utilizar a sintaxe @ para especificar o carimbo de data/hora ou versão como parte do nome da tabela. O carimbo de data/hora deve estar no formato yyyyMMddHHmmssSSS. Você pode especificar uma versão após @ precedendo um v à versão. Consulte o seguinte código para ver um exemplo de sintaxe:

SQL
Python

SQL
SELECT * FROM people10m@20190101000000000
SELECT * FROM people10m@v123

Python
spark.read.table("people10m@20190101000000000")
spark.read.table("people10m@v123")

O que são pontos de verificação de logs de transações?

As versões da tabela são registradas como arquivos JSON dentro do diretório de log de transações, que é armazenado junto com os dados da tabela. Para otimizar a consulta dos pontos de verificação, as versões da tabela são agregadas aos arquivos dos pontos de verificação do Parquet, evitando a necessidade de ler todas as versões do JSON do histórico da tabela. O Databricks otimiza a frequência de pontos de verificação para o tamanho dos dados e a carga de trabalho. Os usuários não devem precisar interagir diretamente com os pontos de verificação. A frequência dos pontos de verificação está sujeita a alterações sem aviso prévio.

Configurar a retenção de dados para consultas de viagem do tempo

Para consultar uma versão anterior da tabela, você deve reter tanto os arquivos de log quanto os dados dessa versão.

Os arquivos de dados são excluídos quando VACUUM é executado em uma tabela. A remoção automática dos arquivos de log é gerenciada após verificar as versões da tabela.

Como a maioria das tabelas tem VACUUM executado regularmente, as consultas pontuais devem respeitar o limite de retenção para VACUUM, que é de sete dias por default.

Para aumentar o limite de retenção de dados para tabelas, é necessário configurar as seguintes propriedades da tabela:

delta.logRetentionDuration = "interval <interval>": controla por quanto tempo o histórico de uma tabela é mantido. O padrão é interval 30 days.
delta.deletedFileRetentionDuration = "interval <interval>": determina que o limite que o VACUUM utiliza para remover arquivos de dados não é mais referenciado na versão da tabela atual. O padrão é interval 7 days.

Você pode especificar as propriedades da tabela durante a criação da tabela ou defini-las com uma instrução ALTER TABLE. Consulte Referência de propriedades da tabela.

nota

No Databricks Runtime 18.0 e acima, logRetentionDuration deve ser maior ou igual a deletedFileRetentionDuration. Para tabelas gerenciadas do Unity Catalog, isso se aplica ao Databricks Runtime 12.2 e acima.

Para acessar 30 dias de dados históricos, configure delta.deletedFileRetentionDuration = "interval 30 days" (que corresponde à configuração default para delta.logRetentionDuration).

Aumentar o limite de retenção de dados pode fazer com que seus custos de armazenamento subam, pois mais arquivos de dados são mantidos.

Restaurar uma tabela para um estado anterior

Você pode restaurar uma tabela para seu estado anterior usando o comando RESTORE. Tabelas mantêm internamente versões históricas que permitem que elas sejam restauradas para um estado anterior. Uma versão correspondente ao estado anterior ou um carimbo de data/hora de quando o estado anterior foi criado são compatíveis como opções pelo comando RESTORE .

importante

Você pode restaurar uma tabela já restaurada.
Você pode restaurar uma tabela clonada.
Você deve ter permissão para MODIFY na tabela que está sendo restaurada.
Você não pode restaurar uma tabela para uma versão mais antiga onde os arquivos de dados foram excluídos manualmente ou pelo vacuum. A restauração para essa versão ainda é possível parcialmente se spark.sql.files.ignoreMissingFiles estiver definido como true.
O formato do carimbo de data/hora para restaurar para um estado anterior é yyyy-MM-dd HH:mm:ss. Também há compatibilidade com o fornecimento de apenas uma string de data (yyyy-MM-dd).

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

Para obter detalhes da sintaxe, consulte RESTORE.

importante

A restauração é considerada uma operação de alteração de dados. As entradas de log adicionadas pelo comando RESTORE contêm dataChange definido como true. Se houver um aplicativo downstream, como um Job de transmissão estruturada que processa as atualizações para uma tabela, as entradas do log de alteração de dados adicionadas pela operação de restauração são consideradas como novas atualizações de dados, e o processamento delas pode resultar em dados duplicados.

Por exemplo:

Versão da tabela	Operação	Atualizações de log	Registros em atualizações de log de alterações de dados
0	INSERT	AddFile(/path/to/file-1, dataChange = true)	(name = Viktor, age = 29, (name = George, age = 55)
1	INSERT	AddFile(/path/to/file-2, dataChange = true)	(name = George, age = 39)
2	OPTIMIZE	AddFile(/path/to/file-3, dataChange = false), RemoveFile(/path/to/file-1), RemoveFile(/path/to/file-2)	(Sem registros, pois a compactação do Optimize não altera os dados na tabela)
3	RESTORE(version=1)	RemoveFile(/path/to/file-3), AddFile(/path/to/file-1, dataChange = true), AddFile(/path/to/file-2, dataChange = true)	(name = Viktor, age = 29), (name = George, age = 55), (name = George, age = 39)

No exemplo anterior, o comando RESTORE resulta em atualizações que já foram vistas ao ler a versão 0 e 1 da tabela. Se uma query de transmissão estiver lendo essa tabela, esses arquivos serão considerados como dados recém-adicionados e serão processados novamente.

Restaurar métricas

RESTORE informa as seguintes métricas como um DataFrame de uma única linha quando a operação é concluída:

table_size_after_restore: O tamanho da tabela após a restauração.
num_of_files_after_restore: O número de arquivos na tabela após a restauração.
num_removed_files: Número de arquivos removidos (excluídos logicamente) da tabela.
num_restored_files: número de arquivos restaurados devido à reversão.
removed_files_size: Tamanho total em bytes dos arquivos removidos da tabela.
restored_files_size: Tamanho total em bytes dos arquivos restaurados.

Exemplos de uso da viagem do tempo

Corrigir exclusões acidentais em uma tabela para o usuário: 111

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

Corrigir atualizações incorretas acidentais em uma tabela:

SQL
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 *

Consultar o número de novos clientes adicionados na última semana.

SQL
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

Como localizar a última versão do commit na sessão do Spark?

Para obter o número da versão do último commit gravado pelo SparkSession atual em todos os threads e todas as tabelas, consulte a configuração SQL spark.databricks.delta.lastCommitVersionInSession.

nota

Para tabelas Apache Iceberg, utilize spark.databricks.iceberg.lastCommitVersionInSession em vez de spark.databricks.delta.lastCommitVersionInSession.

SQL
Python
Scala

SQL
SET spark.databricks.delta.lastCommitVersionInSession

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

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

Se nenhum commit tiver sido feito pelo SparkSession, consultar a chave retornará um valor vazio.

nota

Se o mesmo SparkSession for compartilhado em vários threads, isso é semelhante ao compartilhamento de uma variável em vários threads; poderão surgir condições de corrida, visto que o valor da configuração é atualizado simultaneamente.

Recuperar história de uma tabela​

Esquema de história​

Compreendendo partitionBy em parâmetros de operação​

Exemplo​

Métricas de operações​

O que é viagem do tempo?​

Sintaxe da viagem do tempo​

O que são pontos de verificação de logs de transações?​

Configurar a retenção de dados para consultas de viagem do tempo​

Restaurar uma tabela para um estado anterior​

Restaurar métricas​

Exemplos de uso da viagem do tempo​

Como localizar a última versão do commit na sessão do Spark?​

Recuperar história de uma tabela

Esquema de história

Compreendendo `partitionBy` em parâmetros de operação

Exemplo

Métricas de operações

O que é viagem do tempo?

Sintaxe da viagem do tempo

O que são pontos de verificação de logs de transações?

Configurar a retenção de dados para consultas de viagem do tempo

Restaurar uma tabela para um estado anterior

Restaurar métricas

Exemplos de uso da viagem do tempo

Como localizar a última versão do commit na sessão do Spark?