Trabalhar com o histórico de tabelas do Delta Lake

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

Recuperar o histórico da tabela Delta

Você pode recuperar informações de operações, usuário, carimbo de data/hora e assim por diante relativas a cada gravação em uma tabela do Delta por meio do comando history. As operações são retornadas em ordem cronológica inversa.

A retenção do histórico de tabelas é determinada pela configuração da tabela delta.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 de sintaxe do Spark SQL, confira DESCRIBE HISTORY.

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

O Explorador de Catálogos fornece uma exibição visual dessas informações detalhadas da tabela e do histórico de tabelas Delta. Além do esquema de tabela e dos dados de exemplo, você pode clicar na guia Histórico para ver o histórico da tabela que é exibido com DESCRIBE HISTORY.

Esquema de histórico

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

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|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...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

Chaves de métricas da operação

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 definições essenciais do mapa por operação.

O que é a viagem no tempo do Delta Lake?

O Delta Lake 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:

• Recriando análises, relatórios ou saídas (por exemplo, a saída de um modelo de aprendizado de máquina). 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 do Delta

Você consulta uma tabela do Delta 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 carimbo de data/hora ou data 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 após @, incluindo v no início da versão. Confira o código a seguir para obter uma sintaxe de exemplo:

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 log de transações?

O Delta Lake registra versões de tabela como arquivos JSON no diretório _delta_log, que é armazenado juntamente com os dados da tabela. Para otimizar a consulta de ponto de verificação, o Delta Lake agrega versões de tabela aos arquivos de ponto de verificação Parquet, eliminando a necessidade de ler todas as versões JSON do histórico de tabelas. O Azure Databricks otimiza a frequência de ponto de verificação para o tamanho dos dados e a carga de trabalho. Os usuários não devem interagir diretamente com pontos de verificação. A frequência do ponto de verificação está sujeita a alterações sem aviso prévio.

Configurar retenção de dados para consultas de viagem no tempo

Para consultar uma versão anterior da tabela, você precisa manter tanto o log quanto os arquivos de dados para essa versão.

Os arquivos de dados são excluídos quando VACUUM é executado em uma tabela. O Delta Lake gerencia a remoção de arquivo de log automaticamente após versões de tabela de ponto de verificação.

Como a maioria das tabelas Delta têm VACUUM executadas nelas regularmente, as consultas pontuais devem respeitar o limite de retenção para VACUUM, que é de sete dias por padrão.

Para aumentar o limite de retenção de dados para tabelas Delta, você deve 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 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.

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

Restaurar um estado anterior de uma tabela Delta

Você pode restaurar uma tabela Delta para seu estado anterior com o RESTORE comando. As tabela do Delta mantêm internamente as versões históricas que permitem a restauração para um estado anterior. Uma versão correspondente ao estado anterior ou um carimbo de data/hora de quando o estado anterior foi criado têm suporte como opções pelo comando RESTORE.

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

Para obter detalhes da sintaxe, confira RESTORE.

Restaurar métricas

RESTORE relata as seguintes métricas como um DataFrame de linha única 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 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.

  

Exemplos de como usar a viagem no tempo do Delta Lake

• 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
  

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

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

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

Como encontrar a versão do último commit na sessão do Spark?

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.delta.lastCommitVersionInSession.

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Se o SparkSession não tiver feito nenhum commit, a consulta à chave irá retornar um valor vazio.