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. No Azure Databricks, os conflitos de gravação dependem do nível de isolamento.

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

• Vários gravadores em vários clusters podem modificar simultaneamente uma partição de tabela. Os escritores veem uma vista instantânea consistente da tabela e as gravações ocorrem em uma ordem sequencial.
  • 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.

Consulte Quais são as 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);

Conflitos de gravação com concorrência ao nível de linha

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

A simultaneidade em nível de linha está geralmente disponível no Databricks Runtime 14.2 e superior. A simultaneidade em nível de linha é suportada por padrão para as seguintes condições:

• Tabelas com vetores de exclusão habilitados e sem particionamento.
• Tabelas com agrupamento líquido, a menos que você tenha desabilitado vetores de exclusão.

As tabelas com partições não suportam simultaneidade em 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 para simultaneidade em nível de linha.

Para outras versões do Databricks Runtime, consulte Comportamento de pré-visualização da concorrência ao nível da linha (legado).

MERGE INTO O suporte para a simultaneidade a nível de linha requer o Photon no Databricks Runtime 14.2. No Databricks Runtime 14.3 LTS e superior, 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 com simultaneidade em nível de linha habilitada.

Escrever conflitos sem simultaneidade em 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 suportam simultaneidade em 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 em nível de linha.

Limitações para simultaneidade em nível de linha

Aplicam-se algumas limitações à simultaneidade a nível de linha. Para as operações seguintes, a resolução de conflitos segue o padrão normal de simultaneidade para conflitos de escrita no Azure Databricks. Consulte Escrever conflitos sem concorrência ao nível da linha.

• Comandos com cláusulas condicionais complexas, incluindo o seguinte:
  • Condições em tipos de dados complexos, como structs, arrays ou mapas.
  • Condições usando expressões não determinísticas e subconsultas.
  • Condições que contêm subconsultas correlacionadas.
• No Databricks Runtime 14.2, MERGE comandos devem usar um predicado explícito na tabela de destino para filtrar linhas correspondentes à tabela de origem. Para resolução de mesclagem, o filtro verifica apenas as linhas que podem entrar em conflito com base nas condições do filtro em operações simultâneas.

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

Quando é que o Delta Lake confirma sem ler a tabela?

As operações Delta Lake INSERT ou append não leem o estado da tabela antes de se comprometerem se as seguintes condições forem satisfeitas:

1. A lógica é expressa usando INSERT lógica SQL ou modo de acréscimo.
2. A lógica não contém subconsultas ou condicionais que façam 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.

Escrever níveis de isolamento serializáveis vs. serializáveis

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 oferece suporte a dois níveis de isolamento: Serializable e WriteSerializable.

• Serializável: 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 uma de cada vez que gere o mesmo resultado visto na tabela. Para as operações de gravação, a sequência serial é exatamente a mesma vista no histórico da tabela.

• WriteSerializable (Default): Um nível de isolamento mais fraco do que Serializable. Isto garante apenas que as operações de gravação, isto é, não de leitura, sejam serializáveis. No entanto, isso ainda é mais forte do que o isolamento de Snapshot. WriteSerializable é o nível de isolamento padrão porque fornece um grande equilíbrio de consistência e disponibilidade de dados para a maioria das operações comuns.

  Neste modo, o conteúdo da tabela Delta pode ser diferente do que é esperado da sequência de operações vista no histórico da tabela. Isso ocorre porque esse modo permite que certos pares de gravações simultâneas (digamos, operações X e Y) prossigam de tal forma que o resultado seria como se Y fosse executado antes de X (ou seja, serializável entre eles), mesmo que o histórico mostrasse que Y foi cometido depois de X. Para não permitir essa reordenação, defina o nível de isolamento da tabela como Serializável para fazer com que essas transações falhem.

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

Para o nível Serializável, um leitor sempre vê apenas tabelas que estão de acordo com o histórico. Para o nível WriteSerializable, um leitor pode ver uma tabela que não existe no Delta log.

Por exemplo, considere o cenário em que uma transação de exclusão de longa duração e uma transação de inserção começam ao mesmo tempo e a versão v0de leitura. A transação insert confirma primeiro e cria a versão v1. Depois disso, a transação de exclusão tenta confirmar v2:

t0: deleteTxn_START
t1: insertTxn_START
t2: insertTxn_COMMIT(v1)
t3: deleteTxn_COMMIT(v2)

Neste cenário, deleteTxn não viu os dados inseridos insertTxn pelo e por isso não os excluiu:

• Sob Serializable isolamento, deleteTxn não é permitido cometer, e ocorre um conflito.
• Sob WriteSerializable isolamento, deleteTxn é permitido confirmar porque as transações podem ser ordenadas. O estado resultante da tabela é como se insertTxn tivesse ocorrido após deleteTxn, portanto, as linhas inseridas fazem parte da tabela. No entanto, o histórico Delta mostra a ordem de confirmação física e que insertTxn (v1) aconteceu antes do deleteTxn (v2).

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

Definir o nível de isolamento

Você define 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')

Evite conflitos usando condições de particionamento e comando separado

Em todos os casos marcados como "pode entrar em conflito", se as duas operações entrarão em conflito depende se elas operam no mesmo conjunto de arquivos. Você pode separar os dois conjuntos de arquivos 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' entrarã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, particionar uma tabela de acordo com as condições comumente usadas no comando pode reduzir os conflitos significativamente. No entanto, particionar uma tabela por uma coluna que tem alta cardinalidade pode levar a outros problemas de desempenho devido ao grande número de subdiretórios.

Exceções de conflitos

Quando ocorrer um conflito de transações, 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 INSERT, DELETE, UPDATE, ou MERGE operações.

Com o nível de isolamento padrão de WriteSerializable, os ficheiros adicionados por operações cegasINSERT (ou seja, operações que acrescentam dados sem ler qualquer dado) não conflitam com nenhuma operação, mesmo que toquem na mesma partição (ou em qualquer lugar numa tabela não particionada). Se o nível de isolamento estiver definido como Serializable, as inserções cegas podem entrar em conflito.

Importante: Os apêndices cegos podem conflitar no modo WriteSerializable se várias transações simultâneas DELETE, UPDATEou operações MERGE estiverem em execução e puderem fazer referência a valores inseridos por apêndices cegos. Para evitar esse conflito, siga um destes procedimentos:

• Certifique-se de que as operações simultâneas DELETE, UPDATEou MERGE não leiam os dados anexados.
• Tenha no máximo uma operação DELETE, UPDATEou MERGE que possa ler os dados anexados.

Essa exceção geralmente é lançada durante operações simultâneas DELETE, UPDATEou MERGE . Embora as operações simultâneas possam estar atualizando fisicamente diferentes diretórios de partição, um deles pode ler a mesma partição que o outro atualiza simultaneamente, causando assim um conflito. Pode evitar esta situação ao tornar a separação explícita na condição de operação. Considere o seguinte exemplo.

// 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 executa o código acima em simultâneo para diferentes datas ou países. Como cada trabalho está a funcionar numa partição independente da tabela Delta de destino, não se espera qualquer conflito. No entanto, a condição não é explícita o suficiente e pode verificar toda a tabela e pode entrar em conflito com operações simultâneas atualizando quaisquer outras partições. Em vez disso, pode reescrever a sua declaração para adicionar uma data e um país específicos à condição de intercalação, conforme mostrado no exemplo seguinte.

// 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()

Agora é seguro executar esta operação em simultâneo em datas e países diferentes.

ConcurrentDeleteReadException

Essa exceção ocorre quando uma operação simultânea excluiu um arquivo lido pela operação. As causas comuns são um DELETE, UPDATEou MERGE operação que reescreve arquivos.

ConcurrentDeleteDeleteException

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

MetadataChangedException

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

Exceção de Transação Concorrente

Se uma consulta de streaming usando o mesmo local de ponto de verificação for iniciada várias vezes simultaneamente e tentar gravar na tabela Delta ao mesmo tempo. Nunca deve ter duas consultas em fluxo contínuo a utilizar a mesma localização de ponto de verificação e serem executadas ao mesmo tempo.

ProtocolChangedException

Esta exceção pode ocorrer nos seguintes casos:

• Quando sua tabela Delta é atualizada para uma nova versão de protocolo. Para que operações futuras sejam bem-sucedidas, talvez seja necessário atualizar seu Databricks Runtime.
• Quando múltiplos escritores estão a criar ou substituir uma tabela ao mesmo tempo.
• Quando vários escritores estão a escrever num caminho vazio ao mesmo tempo.

Consulte Compatibilidade de recursos e protocolos do Delta Lake para obter mais detalhes.

Comportamento de visualização de concorrência em nível de linha (legado)

Esta seção descreve comportamentos de visualização para simultaneidade em nível de linha no Databricks Runtime 14.1 e abaixo. A concorrência a nível de linha sempre requer vetores de eliminação.

No Databricks Runtime 13.3 LTS e versões superiores, as tabelas com clustering líquido ativado habilitam automaticamente a concorrência a nível de linha.

No Databricks Runtime 14.0 e 14.1, você pode habilitar a simultaneidade em 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

Nos Databricks Runtime 14.1 e anteriores, a computação não-Photon suporta apenas a simultaneidade ao nível de linha para operações DELETE.