Níveis de isolamento e conflitos de gravação no Azure Databricks

O nível de isolamento de uma tabela define o grau em que uma transação deve ser isolada das modificações feitas por operações simultâneas. Os conflitos de gravação no Azure Databricks dependem do nível de isolamento.

O Delta Lake fornece garantias de transação ACID entre leituras e gravações. Isso significa que:

• Vários gravadores em vários clusters podem modificar simultaneamente uma partição de tabela. Os gravadores veem uma exibição de instantâneo consistente da tabela e as gravações ocorrem em uma ordem serial.
  • Os leitores continuam a ver uma exibição de instantâneo consistente da tabela com a qual o trabalho do Azure Databricks começou, mesmo quando uma tabela é modificada durante um trabalho.

Confira O que são garantias ACID no Azure Databricks?.

-- Set a table property.
ALTER TABLE table-name SET TBLPROPERTIES ('delta.isolationLevel' = 'Serializable')

-- Enable a feature using a table property and update the table protocol.
ALTER TABLE table_name SET TBLPROPERTIES ('delta.enableDeletionVectors' = true);

-- Drop a table feature.
ALTER TABLE table_name DROP FEATURE deletionVectors;

-- Upgrade to UniForm.
REORG TABLE table_name APPLY (UPGRADE UNIFORM(ICEBERG_COMPAT_VERSION=2));

-- Update the table schema.
ALTER TABLE table_name ADD COLUMNS (col_name STRING);

Gravar conflitos com simultaneidade no nível de linha

A simultaneidade no nível de linha reduz conflitos entre operações de gravação simultâneas, detectando alterações no nível da linha e resolvendo automaticamente conflitos que ocorrem quando gravações simultâneas atualizam ou excluem linhas diferentes no mesmo arquivo de dados.

A simultaneidade no nível da linha está em disponibilidade geral no Databricks Runtime 14.2 e superior. A simultaneidade no nível da linha tem suporte por padrão para as seguintes condições:

• Tabelas com vetores de exclusão ativados e sem particionamento.
• Tabelas com clustering líquido, a menos que você tenha desativado os vetores de exclusão.

Tabelas com partições não dão suporte à simultaneidade no nível de linha, mas ainda podem evitar conflitos entre OPTIMIZE e todas as outras operações de gravação quando os vetores de exclusão estão habilitados. Consulte Limitações de simultaneidade no nível de linha.

Para outras versões do Databricks Runtime, confira Comportamento de visualização de simultaneidade no nível de linha (herdado).

MERGE INTO suporte para simultaneidade de nível de linha requer o Photon no Databricks Runtime 14.2. No Databricks Runtime 14.3 LTS e posteriores, o Photon não é necessário.

A tabela a seguir descreve quais pares de operações de gravação podem entrar em conflito em cada nível de isolamento quando a simultaneidade no nível de linha está habilitada.

Conflitos de gravação sem simultaneidade no nível de linha

A tabela a seguir descreve quais pares de operações de gravação podem entrar em conflito em cada nível de isolamento.

As tabelas não dão suporte à simultaneidade no nível de linha se tiverem partições definidas ou não tiverem vetores de exclusão habilitados. O Databricks Runtime 14.2 ou superior é necessário para simultaneidade no nível de linha.

Limitações de simultaneidade no nível de linha

Algumas limitações se aplicam à simultaneidade no nível de linha. No caso das operações a seguir, a resolução de conflitos segue a simultaneidade normal em conflitos de gravação no Azure Databricks. Consulte Conflitos de gravação sem simultaneidade no nível de linha.

• Comandos com cláusulas condicionais complexas, incluindo o seguinte:
  • Condições em tipos de dados complexos, por exemplo, structs, matrizes ou mapas.
  • Condições que usam subconsultas e expressões não determinísticas.
  • Condições que contêm subconsultas correlacionadas.
• No Databricks Runtime 14.2, os comandos MERGE precisam usar um predicado explícito na tabela de destino para filtrar as linhas que correspondem à tabela de origem. Para a resolução da mesclagem, o filtro verifica apenas as linhas que talvez sejam conflitantes com base nas condições do filtro em operações simultâneas.

Todas as limitações de vetores de exclusão também se aplicam. Confira Limitações.

Quando o Delta Lake confirma sem ler a tabela?

As operações INSERT ou de acréscimo do Delta Lake não leem o estado da tabela antes de confirmar se as seguintes condições foram atendidas:

1. A lógica é expressa usando a lógica do SQL INSERT ou o modo de acréscimo.
2. A lógica não contém subconsultas nem condicionais que fazem referência à tabela direcionada pela operação de gravação.

Como em outras confirmações, o Delta Lake valida e resolve as versões da tabela na confirmação usando metadados no log de transações, mas nenhuma versão da tabela é realmente lida.

Gravar níveis de isolamento write serializable versus serializable

O nível de isolamento de uma tabela define o grau em que uma transação deve ser isolada das modificações feitas por transações simultâneas. O Delta Lake no Azure Databricks dá suporte a dois níveis de isolamento: Serializable e WriteSerializable.

• Serializable: o nível de isolamento mais forte. Ele garante que as operações de gravação confirmadas e todas as leituras sejam serializáveis. As operações são permitidas desde que exista uma sequência serial de execução delas, uma por vez, que gere o mesmo resultado que o visto na tabela. Nas operações de gravação, a sequência serial é exatamente igual à vista no histórico da tabela.

• WriteSerializable (padrão): um nível de isolamento mais fraco que o Serializable. Ele garante a serialização somente das operações de gravação (ou seja, não as de leitura). No entanto, ele ainda é mais forte do que o isolamento do instantâneo. O WriteSerializable é o nível de isolamento padrão, pois fornece um grande equilíbrio de consistência e disponibilidade de dados nas operações mais comuns.

  Nesse modo, o conteúdo da tabela Delta pode ser diferente do esperado da sequência de operações vistas no histórico da tabela. Isso porque esse modo permite que determinados pares de gravações simultâneas (por exemplo, operações X e Y) prossigam, de modo que o resultado seria como se Y tivesse sido executado antes de X (ou seja, serializável entre eles), mesmo que o histórico mostre que Y foi confirmado após X. Para não permitir essa reordenação, defina o nível de isolamento da tabela como Serializable para fazer com que essas transações falhem.

As operações de leitura sempre usam isolamento de instantâneo. O nível de isolamento de gravação determina se é possível ou não que um leitor veja um instantâneo de uma tabela que, de acordo com o histórico, "nunca existiu".

No nível Serializable, um leitor sempre vê apenas tabelas que estão em conformidade com o histórico. No nível WriteSerializable, um leitor pode ver uma tabela que não existe no log de Delta.

Por exemplo, considere txn1, uma exclusão de execução prolongada, e txn2, que insere dados excluídos por txn1. txn2 e txn1 são concluídas e são registradas nessa ordem no histórico. De acordo com o histórico, os dados inseridos em txn2 não devem existir na tabela. No nível Serializable, um leitor nunca verá os dados inseridos por txn2. No entanto, no nível WriteSerializable, um leitor pode, em algum momento, ver os dados inseridos por txn2.

Para obter mais informações sobre os tipos de operações que podem entrar em conflito em cada nível de isolamento e os possíveis erros, confira Evitar conflitos usando particionamento e condições de comando separadas.

Definir o nível de isolamento

Você pode definir o nível de isolamento usando o comando ALTER TABLE.

ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.isolationLevel' = <level-name>)

onde <level-name> é Serializable ou WriteSerializable.

Por exemplo, para alterar o nível de isolamento do padrão WriteSerializable para Serializable, execute:

ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.isolationLevel' = 'Serializable')

Evitar conflitos usando particionamento e condições de comando separadas

Em todos os casos marcados como "pode entrar em conflito", se as duas operações entrarão em conflito ou não depende se elas operam no mesmo conjunto de arquivos. Você pode tornar os dois conjuntos de arquivos não contíguos particionando a tabela pelas mesmas colunas usadas nas condições das operações. Por exemplo, os dois comandos UPDATE table WHERE date > '2010-01-01' ... e DELETE table WHERE date < '2010-01-01' estarão em conflito se a tabela não for particionada por data, pois ambos podem tentar modificar o mesmo conjunto de arquivos. Particionar a tabela por date evitará o conflito. Portanto, o particionamento de uma tabela de acordo com as condições comumente usadas no comando pode reduzir conflitos significativamente. No entanto, o particionamento de uma tabela por uma coluna com alta cardinalidade pode levar a outros problemas de desempenho devido a um grande número de subdiretórios.

Exceções de conflito

Quando ocorrer um conflito de transações, você observará uma das seguintes exceções:

ConcurrentAppendException

Essa exceção ocorre quando uma operação simultânea adiciona arquivos na mesma partição (ou em qualquer lugar em uma tabela não particionada) que sua operação lê. As adições de arquivo podem ser causadas por operações INSERT, DELETE, UPDATE ou MERGE.

Com o nível de isolamento padrão de WriteSerializable, arquivos adicionados por operações cegas INSERT (ou seja, operações que adicionam dados sem ler nenhum dado) não entram em conflito com nenhuma operação, mesmo que afetem a mesma partição (ou qualquer lugar em uma tabela não particionada). Se o nível de isolamento for definido como Serializable, os acréscimos ocultos poderão entrar em conflito.

Essa exceção geralmente é lançada durante operações DELETE, UPDATE ou MERGE simultâneas. Embora as operações simultâneas possam estar atualizando fisicamente diretórios de partição diferentes, um deles pode ler a mesma partição que a outra está atualizando simultaneamente, causando um conflito. Você pode evitar isso tornando a separação explícita na condição da operação. Considere o exemplo a seguir.

// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t").merge(
    source.as("s"),
    "s.user_id = t.user_id AND s.date = t.date AND s.country = t.country")
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

Suponha que você execute o código acima simultaneamente em diferentes datas ou países. Como cada trabalho está trabalhando em uma partição independente na tabela do Delta de destino, você espera que não ocorram conflitos. No entanto, a condição não é explícita o suficiente e pode verificar toda a tabela e entrar em conflito com operações simultâneas atualizando qualquer outra partição. Em vez disso, você pode reescrever sua instrução para adicionar datas e países específicos à condição de mesclagem, conforme mostrado no exemplo a seguir.

// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t").merge(
    source.as("s"),
    "s.user_id = t.user_id AND s.date = t.date AND s.country = t.country AND t.date = '" + <date> + "' AND t.country = '" + <country> + "'")
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

Essa operação agora é segura para ser executada simultaneamente em diferentes datas e países.

ConcurrentDeleteReadException

Essa exceção ocorre quando uma operação simultânea excluiu um arquivo que sua operação leu. As causas comuns são uma operação DELETE, UPDATE, ou MERGE que reescreve arquivos.

ConcurrentDeleteDeleteException

Essa exceção ocorre quando uma operação simultânea excluiu um arquivo que sua operação também exclui. Isso pode ser causado por duas operações de compactação simultâneas reescrevendo os mesmos arquivos.

MetadataChangedException

Essa exceção ocorre quando uma transação simultânea atualiza os metadados de uma tabela do Delta. As causas comuns são operações ou gravações ALTER TABLE em sua tabela do Delta que atualiza o esquema da tabela.

ConcurrentTransactionException

Se uma consulta de streaming que usa o mesmo local de ponto de verificação for iniciada várias vezes simultaneamente e tentar gravar na tabela do Delta ao mesmo tempo. Você nunca deve ter duas consultas de streaming que usam o mesmo local de ponto de verificação e sejam executadas ao mesmo tempo.

ProtocolChangedException

Essa exceção pode ocorrer nos seguintes casos:

• Quando a tabela do Delta é atualizada para uma nova versão do protocolo. Para que as operações futuras possam ser bem-sucedidas, talvez seja necessário atualizar o Databricks Runtime.
• Quando vários gravadores estão criando ou substituindo uma tabela ao mesmo tempo.
• Quando vários gravadores estão gravando em um caminho vazio ao mesmo tempo.

Confira Como o Azure Databricks gerencia a compatibilidade de recursos do Delta Lake? para obter mais detalhes.

Comportamento de pré-visualização de simultaneidade no nível da linha (herdado)

Esta seção descreve os comportamentos de visualização para simultaneidade no nível de linha no Databricks Runtime 14.1 e versões anteriores. A simultaneidade no nível de linha sempre requer vetores de exclusão.

No Databricks Runtime 13.3 LTS e versões posteriores, tabelas com clustering líquido habilitados habilitam automaticamente a simultaneidade no nível de linha.

No Databricks Runtime 14.0 e 14.1, você pode habilitar a simultaneidade no nível de linha para tabelas com vetores de exclusão definindo a seguinte configuração para o cluster ou SparkSession:

spark.databricks.delta.rowLevelConcurrencyPreview = true

No Databricks Runtime 14.1 e versões anteriores, a computação não do Photon só dá suporte à simultaneidade de nível de linha para operações DELETE.