Substituir dados seletivamente com o Delta Lake

O Delta Lake tem as seguintes opções distintas para substituições seletivas:

• A opção replaceWhere substitui atomicamente todos os registros que correspondem a um determinado predicado.
• Você pode substituir diretórios de dados com base em como as tabelas são particionadas usando substituições de partição dinâmica.

Para a maioria das operações, o Databricks recomenda usar replaceWhere para especificar quais dados substituir.

Substituição seletiva arbitrária com replaceWhere

Você pode substituir seletivamente apenas os dados que correspondem a uma expressão arbitrária.

Por exemplo, substitua atomicamente os eventos em janeiro na tabela de destino, que é particionada por start_date, com os dados em replace_data.

Python

(replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")
)

Scala

replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events REPLACE WHERE start_date >= '2017-01-01' AND end_date <= '2017-01-31' SELECT * FROM replace_data

Este código de exemplo grava os dados em replace_data, valida se todas as linhas correspondem ao predicado e executa uma substituição atômica usando a semântica overwrite. Se algum valor na operação estiver fora da restrição, ocorrerá uma falha nessa operação com um erro por padrão.

Você pode alterar esse comportamento para valores overwrite dentro do intervalo do predicado e os registros insert que ficam fora do intervalo especificado. Para fazer isso, desabilite a verificação de restrição definindo spark.databricks.delta.replaceWhere.constraintCheck.enabled como false usando uma das seguintes configurações:

Python

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", False)

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", false)

SQL

SET spark.databricks.delta.replaceWhere.constraintCheck.enabled=false

Comportamento herdado

Se você usar o comportamento herdado de replaceWhere, as consultas corresponderão apenas a predicados em colunas de partição e ignorarão todos os dados fora do intervalo de predicados. O comando a seguir substituiria atomicamente o mês de janeiro na tabela de destino, que é particionada por date, com os dados em df:

Python

(df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")
)

Scala

df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")

Para usar o comportamento herdado, defina o spark.databricks.delta.replaceWhere.dataColumns.enabled sinalizador como false:

Python

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", False)

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", false)

SQL

SET spark.databricks.delta.replaceWhere.dataColumns.enabled=false

Substituições de partições dinâmicas

A sobrescrita dinâmica de partições atualiza apenas as partições em que a gravação confirma novos dados e deixa outras partições inalteradas.

Azure Databricks recomenda REPLACE USING para acionar substituições de partição dinâmica. Esse modo funciona em todos os tipos de computação, incluindo databricks SQL warehouses, computação sem servidor e computação clássica, e não exige que você defina uma configuração de sessão do Spark. Consulte sobrescritas de partição dinâmica com REPLACE USING.

partitionOverwriteMode é um modo herdado para substituições de partição dinâmica que exige que você use a computação clássica e defina uma configuração de sessão do Spark. Não há suporte para ele no DATAbricks SQL ou na computação sem servidor. Consulte substituições de partição dinâmica com partitionOverwriteMode (herdado).

As seções abaixo demonstram como usar cada modo.

Substituições de partição dinâmica com REPLACE USING

O Databricks Runtime 16.3 ou versões posteriores dá suporte a substituições de partições dinâmicas para tabelas usando REPLACE USING. Você pode substituir seletivamente dados em todos os tipos de computação, sem a necessidade de definir uma configuração de sessão do Spark. REPLACE USING permite um comportamento de substituição atômica independente de computação que funciona em SQL warehouses do Databricks, computação sem servidor e computação clássica.

REPLACE USING substitui apenas as partições alvo dos dados de entrada. Todas as outras partições permanecem inalteradas.

REPLACE USING só há suporte no SQL. Para obter detalhes, consulte INSERT na referência da linguagem SQL.

Os exemplos a seguir usam REPLACE USING:

INSERT INTO TABLE events
  REPLACE USING (event_id, start_date)
  SELECT * FROM source_data

Tenha em mente as seguintes restrições e comportamentos em relação às sobreposições dinâmicas de partição:

• No Databricks Runtime 17.2 e superior, há suporte para tabelas particionadas, tabelas não particionadas e tabelas com clustering líquido.
• No Databricks Runtime 16.3 a 17.1, há suporte apenas para tabelas particionadas e você deve especificar o conjunto completo de colunas de partição da tabela na USING cláusula.
• Sempre valide se os dados gravados alcançam apenas as partições esperadas. Uma única linha na partição errada pode acabar sobrescrevendo involuntariamente toda a partição.

Se você precisar de uma lógica de correspondência mais personalizável do que o que REPLACE USING suporta, como tratar os valores de NULL como iguais, use o complementar REPLACE ON. Consulte INSERT para obter detalhes.

Substituições de partição dinâmica com partitionOverwriteMode (herdado)

O Databricks Runtime 11.3 LTS e versões posteriores oferecem suporte a substituições dinâmicas de partição para tabelas particionadas usando o modo de substituição: INSERT OVERWRITE no SQL ou gravação de DataFrame com df.write.mode("overwrite"). Esse tipo de substituição só está disponível para computação clássica, não para databricks SQL warehouses ou computação sem servidor.

Para usar o modo de substituição de partição dinâmica, defina a configuração spark.sql.sources.partitionOverwriteMode de sessão do Spark como dynamic. Como alternativa, você pode definir a opção DataFrameWriterpartitionOverwriteMode como dynamic. Se presente, a opção específica da consulta substitui o modo definido na configuração da sessão. O valor padrão para spark.sql.sources.partitionOverwriteMode é static.

O exemplo a seguir usa partitionOverwriteMode:

SQL

SET spark.sql.sources.partitionOverwriteMode=dynamic;
INSERT OVERWRITE TABLE default.people10m SELECT * FROM morePeople;

Python

(df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")
)

Scala

df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")

Tenha em mente as seguintes restrições e comportamentos para partitionOverwriteMode:

• Você não pode definir overwriteSchema como true.
• Você não pode especificar ambos partitionOverwriteMode e replaceWhere na mesma DataFrameWriter operação.
• Se você especificar uma replaceWhere condição usando uma DataFrameWriter opção, o Delta Lake aplicará essa condição para controlar quais dados são substituídos. Essa opção tem precedência sobre a configuração no nível da partitionOverwriteMode sessão.
• Sempre valide se os dados gravados alcançam apenas as partições esperadas. Uma única linha na partição errada pode acabar sobrescrevendo involuntariamente toda a partição.