Trabalhar com o histórico de tabelas

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

Recuperar histórico da tabela

Execute o DESCRIBE HISTORY comando para recuperar informações incluindo as operações, o utilizador e o carimbo temporal de cada gravação numa tabela. As operações são retornadas em ordem cronológica inversa.

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

A retenção do histórico da tabela é 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 detalhes da sintaxe em Scala, Java e Python, consulte a documentação da API Delta Lake.

O Explorador de Catálogos mostra visualmente o histórico da tabela no separador Histórico .

Identificar o tipo de OPTIMIZE operação

Autocompactação, agrupamento líquido e ordenação Z aparecem todos no histórico da tabela como OPTIMIZE operações. Para determinar qual foi executado, inspecione a coluna operationParameters.

Para classificar todas OPTIMIZE as 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 secções seguintes descrevem cada operationParameters valor em detalhe. Para definições das operationMetrics chaves selecionadas pela consulta anterior, veja Métricas de operação.

Compactação automática

A autocompactação define o auto parâmetro para true. O Azure Databricks desencadeia automaticamente a compactação automática após uma escrita. Quando auto é false, um utilizador ou um trabalho agendado executou o OPTIMIZE comando.

Por exemplo, uma operação de autocompactação mostra o seguinte:

operationParameters: {
  "auto": "true"
}

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

Agrupamento de líquidos

O agrupamento líquido preenche o clusterBy parâmetro com os nomes das colunas de agrupamento. Um array vazio clusterBy ([]) indica apenas compactação de ficheiros.

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

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

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

Ordenação em Z

A ordenação Z preenche o zOrderBy parâmetro com os nomes das colunas da ordem Z. Um array vazio zOrderBy ([]) indica que a operação não aplicou a ordem Z.

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

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

Âmbito da operação

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

• Uma matriz predicate vazia ([]) significa que a operação foi executada na tabela inteira.
• Um array preenchido 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 dirigida às partições que correspondem a year = 2024 mostra o seguinte:

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

Viagem no tempo

A viagem no tempo suporta consultar versões anteriores das tabelas, registadas no registo de transações, com base no timestamp ou na versão da tabela. Você pode usar a viagem no tempo para aplicativos como os seguintes:

• Recriar análises, relatórios ou resultados, como o resultado de um modelo de aprendizagem automática. Isto pode ser útil para depuração ou auditoria, especialmente em indústrias reguladas.
• Escrever consultas temporais complexas.
• Corrigir erros nos seus dados.
• Proporcionar isolamento ao nível de instantâneo para um conjunto de consultas em tabelas de rápida mutação.

Sintaxe da viagem no tempo

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 um:
  • '2018-10-18T22:15:12.013Z', ou seja, uma cadeia de caracteres que pode ser convertida num timestamp
  • 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 num marca temporal
• version é um valor longo que pode ser obtido a partir da saída de DESCRIBE HISTORY table_spec.

Nem timestamp_expression nem version podem ser subconsultas.

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

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 deve estar no formato yyyyMMddHHmmssSSS. Pode especificar uma versão com @v. Consulte o seguinte código para exemplo de sintaxe:

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

Para consultar uma versão anterior da tabela, deve manter tanto o registo como os ficheiros de dados dessa versão:

• Os arquivos de dados são eliminados quando VACUUM é executado numa tabela.
• Os ficheiros de registo são removidos automaticamente após a criação de pontos de verificação das versões da tabela.

Para aumentar o limiar de retenção de dados para tabelas, deve configurar as seguintes propriedades da tabela, substituindo <format> por delta ou iceberg:

• <format>.logRetentionDuration = "interval <interval>": controla por quanto tempo o histórico de uma tabela é mantido. A predefinição é interval 30 days.
  • Em Databricks Runtime 18.0 e superiores, logRetentionDuration deve ser maior ou igual a deletedFileRetentionDuration. Para tabelas geridas pelo Unity Catalog, isto aplica-se ao Databricks Runtime 12.2 e superiores.
• <format>.deletedFileRetentionDuration = "interval <interval>": determina o limite VACUUM usa para remover arquivos de dados que não são mais referenciados na versão atual da tabela. A predefinição é interval 7 days.

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

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

Exemplos de viagens no tempo

Para corrigir eliminações acidentais numa tabela para o utilizador 111:

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

Para corrigir atualizações acidentais e incorretas numa 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 questionar 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 registo de transações

O registo de transações regista as versões das tabelas como ficheiros JSON dentro do diretório do registo de transações, juntamente com os dados das tabelas.

Para otimizar a consulta de checkpoints, as versões das tabelas são agregadas em ficheiros de checkpoint Parquet, o que melhora o desempenho ao evitar a necessidade de ler todas as versões JSON do histórico das tabelas. Os utilizadores não precisam de interagir diretamente com os pontos de controlo.

O Azure Databricks otimiza a frequência de pontos 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 a um estado anterior

Use o RESTORE comando para restaurar uma tabela a uma versão ou carimbo temporal anterior, incluindo nestes cenários:

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

Considere os seguintes requisitos:

• Para restaurar uma tabela, tem de ter a permissão MODIFY para a tabela.
• Depois de os ficheiros de dados serem eliminados, manualmente ou por VACUUM, não podes restaurar uma tabela para uma versão antiga que faça referência a esses ficheiros. A restauração parcial para esta versão ainda é possível se spark.sql.files.ignoreMissingFiles estiver definida como true.
• Para restaurar por carimbo temporal, 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 de sintaxe, consulte RESTORE.

Comportamento de streaming

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

Para cargas de trabalho a jusante, como um trabalho de streaming estruturado que processa as atualizações de uma tabela, as entradas do registo de alterações de dados adicionadas pela operação de restauro são consideradas novas atualizações de dados, e o seu processamento pode resultar em dados duplicados.

Por exemplo:

No exemplo anterior, o RESTORE comando resulta em atualizações que já tinham sido vistas ao ler as versões 0 e 1 da tabela. Se uma consulta de streaming ler esta tabela novamente, então estes ficheiros são considerados dados recém-adicionados e são processados novamente.

Restaurar métricas

Após concluir, RESTORE reporta as seguintes métricas como uma única linha de DataFrame:

• 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 ficheiros removidos (logicamente eliminados) da tabela.

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

• removed_files_size: Tamanho total, em bytes, dos ficheiros que são removidos da tabela.

• restored_files_size: Tamanho total em bytes dos arquivos que são restaurados.

  

Encontre a última versão de commit

Para obter o número da versão do último commit feito pelo atual SparkSession em todos os threads e em todas as tabelas, execute uma consulta na configuração SQL spark.databricks.<format>.lastCommitVersionInSession. Substitua <format> por delta ou iceberg, consoante o formato da sua 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 pelo SparkSession, consultar a chave retornará um valor vazio.