Trabalhar com o histórico de tabelas

Para tabelas Apache Iceberg e Delta Lake, cada operação que modifica uma tabela cria uma nova versão da tabela. Use informações de histórico para auditar operações, reverter uma tabela ou consultar uma tabela em um ponto específico no tempo usando a viagem no tempo.

Recuperar o histórico da tabela

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

Para as colunas que DESCRIBE HISTORY retornam, os valores na operationParameters coluna e as métricas por operação na operationMetrics coluna, veja Esquema histórico de tabela e métricas de operação.

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

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 da sintaxe do Spark SQL, consulte DESCRIBE HISTORY.

Para obter detalhes de sintaxe scala, Java e Python, consulte a documentação da API delta lake.

O Gerenciador de Catálogos mostra o histórico da tabela visualmente na guia Histórico .

Identificar o tipo de OPTIMIZE operação

Compactação automática, clusterização líquida e ordenação Z aparecem no histórico da tabela como operações OPTIMIZE. Para determinar qual deles foi executado, examine a coluna operationParameters.

Para classificar todas as OPTIMIZE operações no histórico de uma tabela, execute o seguinte:

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;

As seções a seguir descrevem cada operationParameters valor em detalhes. Para definições das operationMetrics chaves selecionadas pela consulta anterior, veja Métricas de operação.

Compactação automática

A compactação automática define o auto parâmetro como true. Azure Databricks dispara a compactação automática automaticamente após uma gravação. Quando auto é false, um usuário ou trabalho agendado executou o OPTIMIZE comando.

Por exemplo, uma operação de compactação automática mostra o seguinte:

operationParameters: {
  "auto": "true"
}

Para obter mais informações sobre compactação automática, consulte Compactação automática.

Agrupamento de líquidos

O liquid clustering preenche o parâmetro clusterBy com os nomes das colunas de agrupamento. Uma matriz vazia clusterBy ([]) indica apenas compactação de arquivo.

Por exemplo, uma operação que agrupou dados pelas colunas date e region mostra o seguinte:

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

Para obter mais informações sobre agrupamento líquido, consulte Usar agrupamento líquido para tabelas.

ordenação em Z

A ordenação Z preenche o zOrderBy parâmetro com os nomes de coluna de ordem Z. Uma matriz vazia zOrderBy ([]) indica que a operação não aplicou a ordenação Z.

Por exemplo, uma operação que aplicou a ordenação Z na date coluna mostra o seguinte:

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

Escopo de operação

O predicate parâmetro indica se a operação foi executada na tabela completa ou apenas parte dela:

• Uma matriz vazia predicate ([]) significa que a operação foi executada em toda a tabela.
• Uma matriz populada predicate significa que um comando direcionado OPTIMIZE table_name WHERE <partition_predicate> foi executado apenas nas partições que correspondem ao predicado.

Por exemplo, uma operação direcionada às partições correspondentes a year = 2024 mostra o seguinte:

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

Viagem no tempo

A viagem no tempo dá suporte à consulta de versões de tabela anteriores com base no carimbo de data/hora ou na versão da tabela (conforme registrado no log de transações). Você pode usar viagens no tempo para aplicativos como o seguinte:

• Recriar análises, relatórios ou saídas, como a saída de um modelo de machine learning. Isso pode ser útil para depuração ou auditoria, especialmente em setores regulamentados.
• Escrevendo consultas temporais complexas.
• Corrigindo erros nos seus dados.
• Fornecimento de isolamento de instantâneos para um conjunto de consultas de tabelas que são alteradas rapidamente.

Sintaxe de viagem no tempo

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

• timestamp_expression pode ser qualquer uma das seguintes opções:
  • '2018-10-18T22:15:12.013Z', ou seja, uma cadeia de caracteres 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 cadeia de caracteres 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.

timestamp_expression e version não podem ser subconsultas.

Somente cadeias de caracteres de data ou carimbo de data/hora são aceitas. Por exemplo, "2019-01-01" e "2019-01-01T00:00:00.000Z". Confira o código a seguir para obter uma sintaxe de exemplo:

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 usar a sintaxe @ para especificar o carimbo de data/hora ou a versão como parte do nome da tabela. O carimbo de data/hora precisa estar no formato yyyyMMddHHmmssSSS. Você pode especificar uma versão com @v. Confira o código a seguir para obter uma sintaxe de exemplo:

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 retenção de dados para consultas de viagem no tempo

Para consultar uma versão anterior da tabela, você deve manter ambos o log e os arquivos de dados para essa versão:

• Os arquivos de dados são excluídos quando VACUUM é executado contra uma tabela.
• Os arquivos de log são removidos automaticamente após versões da tabela de ponto de verificação.

Para aumentar o limite de retenção de dados para tabelas, você deve configurar as seguintes propriedades de tabela, substituindo <format> por:deltaiceberg

• <format>.logRetentionDuration = "interval <interval>": controla por quanto tempo o histórico de uma tabela é mantido. O padrão é interval 30 days.
  • No Databricks Runtime 18.0 e superior, logRetentionDuration deve ser maior ou igual a deletedFileRetentionDuration. Para tabelas gerenciadas do Unity Catalog, isso se aplica a partir da versão 12.2 do Databricks Runtime e superior.
• <format>.deletedFileRetentionDuration = "interval <interval>": determina o limite VACUUM usado para remover arquivos de dados que não são mais referenciados na versão atual da tabela. O padrão é interval 7 days.

Por exemplo, para acessar 30 dias de dados históricos, defina delta.deletedFileRetentionDuration = "interval 30 days", que corresponde à configuração padrão para delta.logRetentionDuration.

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

Exemplos de viagem no tempo

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

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

Para corrigir atualizações incorretas acidentais em uma tabela:

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 o número de novos clientes adicionados na ú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

Pontos de verificação do log de transações

O log de transações registra as versões da tabela como arquivos JSON no diretório de log de transações junto com os dados da tabela.

Para otimizar a consulta dos checkpoints, as versões da tabela são consolidadas em arquivos de checkpoint em Parquet, o que melhora o desempenho ao eliminar a necessidade de ler todas as versões JSON do histórico da tabela. Os usuários não precisam interagir diretamente com pontos de verificação.

O Azure Databricks otimiza a frequência de ponto de verificação para o tamanho dos dados e a carga de trabalho. A frequência do ponto de verificação está sujeita a alterações sem aviso prévio.

Restaurar uma tabela para um estado anterior

Use o comando RESTORE para restaurar uma tabela para uma versão anterior ou para um carimbo de data/hora específico, inclusive nos seguintes cenários:

• Você pode restaurar uma tabela já restaurada.
• Você pode restaurar uma tabela clonada.

Considere os seguintes requisitos:

• Para restaurar uma tabela, você deve ter MODIFY permissão para a tabela.
• Depois que os arquivos de dados são excluídos, manualmente ou por VACUUM, você não pode restaurar uma tabela para uma versão mais antiga que referencia esses arquivos. A restauração parcial para essa versão ainda é possível se spark.sql.files.ignoreMissingFiles estiver definido como true.
• Para restaurar por timestamp, use os formatos yyyy-MM-dd HH:mm:ss ou yyyy-MM-dd.

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.

Comportamento de streaming

A restauração é uma operação de alteração de dados e pode resultar em dados duplicados para cargas de trabalho downstream. As entradas de log adicionadas pelo RESTORE comando contêm dataChange definido como true.

Para cargas de trabalho downstream, como um trabalho de Structured streaming que processa as atualizações em uma tabela, as entradas do log de alterações de dados adicionadas pela operação de restauração são consideradas novas atualizações de dados, e o processamento delas pode resultar em dados duplicados.

Por exemplo:

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

Restaurar métricas

Depois de concluir, RESTORE relata as seguintes métricas como um DataFrame de linha única:

• table_size_after_restore: o tamanho da tabela após a restauração.

• num_of_files_after_restore: o número de arquivos da tabela após a restauração.

• num_removed_files: o número de arquivos removidos (logicamente excluídos) da tabela.

• num_restored_files: o número de arquivos restaurados devido à reversão.

• removed_files_size: o tamanho total em bytes dos arquivos removidos da tabela.

• restored_files_size: o tamanho total em bytes dos arquivos restaurados.

  

Localize a última versão do commit

Para ter acesso ao número da versão do último commit gravado pelo atual SparkSession em todos os threads e todas as tabelas, confira a configuração do SQL spark.databricks.<format>.lastCommitVersionInSession. Substitua <format> por um delta ou icebergoutro, dependendo do formato da tabela.

Por exemplo:

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Se nenhuma confirmação tiver sido feita pela SparkSession, a consulta da chave retornará um valor vazio.