Substituir dados seletivamente com o Delta Lake

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

Opção Caso de uso Tipos de computação com suporte Versão mínima
REPLACE WHERE Sobrescreva atomicamente linhas que atendem a um predicado. Use para substituições com uma condição de correspondência fixa, como colA = 5 ou int_col IN (1, 2, 3). Todos os tipos de computação. SQL no Databricks Runtime 12.2 LTS e versões superiores. Python e Scala no Databricks Runtime 9.1 LTS e versões posteriores.
REPLACE USING Sobrescrita de dados dinâmicos. Substitui todas as linhas que correspondem às colunas especificadas, com base na comparação de igualdade dos valores de coluna, no conjunto de dados fornecido. Todos os tipos de computação. SQL no Databricks Runtime 16.3 e posteriores. Python e Scala no Databricks Runtime 18.2 e posteriores.
REPLACE ON Sobrescrita de dados dinâmicos por expressão booleana. Use para substituições com uma condição de correspondência complexa ou segura para NULL, como s.colA <=> t.colA AND s.colB <=> t.colB. Todos os tipos de computação. SQL no Databricks Runtime 17.1 e posteriores. Python e Scala no Databricks Runtime 18.2 e posteriores.
partitionOverwriteMode Sobrescrita dinâmica de partição legada, que sobrescreve todos os dados existentes em cada partição na qual a operação de gravação gravará novos dados. Não recomendado para novas cargas de trabalho. O SQL dá suporte somente à computação clássica. Python e Scala dão suporte a todos os tipos de computação. SQL, Python e Scala no Databricks Runtime 11.3 LTS e versões posteriores.

Para a maioria dos casos de uso, o Databricks recomenda usar REPLACE USING ou REPLACE WHERE. Use REPLACE ON somente se o seu caso de uso exigir condições de correspondência complexas ou seguras para NULL.

Para obter detalhes sobre o comportamento de substituição de cada opção, consulte INSERT. Para obter uma lista completa das opções do DataFrameWriter Delta Lake, consulte Delta Lake e Apache Iceberg.

Em Scala e Python, replaceOn e replaceUsing não podem ser usados em combinação com replaceWhere, partitionOverwriteMode ou overwriteSchema.

Para consultas de fonte vazias, ambos REPLACE USING e REPLACE ON não excluem dados; no entanto, REPLACE WHERE pode excluir dados.

Important

Se os dados tiverem sido substituídos acidentalmente, você poderá usar restaurar para desfazer a alteração.

REPLACE WHERE

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

Important

Para se beneficiar da atualização incremental ao executar REPLACE WHERE, use fluxos de REPLACE WHERE no Spark Declarative Pipelines (SDP). Consulte processamento em lote com fluxos REPLACE WHERE.

Para substituir atomicamente os eventos de janeiro na tabela de destino, que é particionada por start_date, pelos 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 do predicado, essa operação falhará com um erro por padrão.

Na computação clássica, para alterar esse comportamento para overwrite valores dentro do intervalo do predicado e insert registros fora do intervalo especificado, remova a verificação de restrição definindo spark.databricks.delta.replaceWhere.constraintCheck.enabled como false:

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

Note

REPLACE WHERE aceita um boolean_expression com algumas restrições. Consulte a referência da linguagem SQL em INSERT.

Para consultas de origem vazias, REPLACE WHERE pode excluir linhas de tabela.

Comportamento herdado

Herdado replaceWhere só está disponível na computação clássica. Confira a visão geral da computação clássica.

Se você usar o comportamento herdado de replaceWhere, as consultas sobrescrevem os dados que correspondem a um predicado somente em colunas de partição. 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 spark.databricks.delta.replaceWhere.dataColumns.enabled 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 dados dinâmicos

As sobrescritas de dados dinâmicos substituem seletivamente os dados que correspondem às colunas-chave especificadas ou à expressão booleana, deixando todos os outros dados inalterados. Há suporte para tabelas particionadas, tabelas não particionadas e tabelas com clustering líquido.

Substituições de partição dinâmica são um subconjunto de comportamento de substituição de dados dinâmicos. As sobrescritas dinâmicas de partição substituem todos os dados existentes em cada partição na qual a gravação efetivará novos dados e deixam todas as outras partições inalteradas. Há suporte apenas para tabelas particionadas.

REPLACE USING

SQL com suporte no Databricks Runtime 16.3 e superior. Python e Scala com suporte no Databricks Runtime 18.2 e superior. Para diferenças de comportamento no Databricks Runtime 16.3 a 17.1, consulte o comportamento herdado.

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 não exige que você defina uma configuração de sessão do Spark.

REPLACE USING substitui linhas quando os valores das colunas especificadas são iguais. Todos os outros dados permanecem inalterados.

Use a sobrescrita de dados dinâmicos com REPLACE USING:

Python

(sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")
)

Scala

sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")

SQL

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

Para consultas de origem vazias, REPLACE USING não exclui nenhuma linha de tabela.

Para condições de correspondência complexas ou nulas, use REPLACE ON em vez disso. Consulte REPLACE ON.

Consulte a referência da linguagem SQL em INSERT.

Comportamento herdado

No Databricks Runtime 16.3 a 17.1, REPLACE USING usa o comportamento herdado e permite apenas substituições de partição dinâmica, enquanto o Databricks Runtime 17.2 e superior permite substituições de dados dinâmicos.

Tenha em mente as seguintes restrições e comportamentos para o REPLACE USING comportamento legado:

  • Você deve especificar o conjunto completo das 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.

REPLACE ON

SQL compatível com Databricks Runtime 17.1 e versões posteriores. Python e Scala com suporte no Databricks Runtime 18.2 e superior.

REPLACE ON substitui as linhas quando elas atendem a uma condição definida pelo usuário, ao contrário de REPLACE USING, que substitui as linhas quando as colunas especificadas são consideradas iguais pelo operador de igualdade. Use REPLACE ON quando precisar de uma lógica correspondente que REPLACE USING não dê suporte, como tratar NULL valores como iguais.

Opcionalmente, use a opção targetAlias para especificar um alias para a tabela de destino e as .as() APIs ou .alias() para especificar um alias para os dados de origem.

Para sintaxe SQL, consulte INSERT.

Python

(sourceDataDF.alias("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")
)

Scala

sourceDataDF.as("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events AS t
  REPLACE ON (s.event_id <=> t.event_id AND s.start_date <=> t.start_date)
  (SELECT * FROM source_data) AS s

Para consultas de origem vazias, REPLACE ON não exclui nenhuma linha de tabela.

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

Important

Esse recurso está em Visualização Pública.

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.

Aviso

Quando possível, use INSERT REPLACE USING em vez de sobrescrever a partição com INSERT OVERWRITE PARTITION e spark.sql.sources.partitionOverwriteMode=dynamic. A substituição de partição pode usar dados antigos quando a partição muda.

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.
  • Last updated on