Esquemas de tabela de atualização com evolução do esquema

As tabelas dão suporte à evolução do esquema, permitindo modificações na estrutura da tabela à medida que os requisitos de dados são alterados. Há suporte aos seguintes tipos de alterações:

Faça essas alterações explicitamente usando DDL ou implicitamente usando DML.

Importante

As atualizações no esquema conflitam com todas as operações simultâneas de gravação. A Databricks recomenda coordenar as alterações de esquema para evitar conflitos de gravação.

A atualização de um esquema de tabela encerra todos os fluxos de leitura dessa tabela. Para continuar o processamento, reinicie o fluxo usando os métodos descritos nas considerações de produção para Streaming Estruturado.

Alterações manuais de esquema

Use ALTER TABLE instruções para alterar explicitamente o esquema de uma tabela sem gravar novos dados.

Adicionar colunas

Use ALTER TABLE ... ADD COLUMNS para adicionar uma ou mais colunas a uma tabela existente, opcionalmente especificando a posição e um comentário:

ALTER TABLE table_name ADD COLUMNS (col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

Por padrão, a nulabilidade é true.

Exemplo: adicionar campos aninhados

A adição de colunas aninhadas tem suporte apenas para structs. Não há suporte a matrizes e mapas.

Para adicionar uma coluna a um campo aninhado, use:

ALTER TABLE table_name ADD COLUMNS (col_name.nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

Por exemplo, se o esquema anterior à execução de ALTER TABLE boxes ADD COLUMNS (colB.nested STRING AFTER field1) for:

- root
| - colA
| - colB
| +-field1
| +-field2

o esquema após será:

- root
| - colA
| - colB
| +-field1
| +-nested
| +-field2

Alterar comentários das colunas e a ordem

Use ALTER TABLE ... ALTER COLUMN para atualizar o comentário de uma coluna ou reordená-lo em relação a outras colunas:

ALTER TABLE table_name ALTER [COLUMN] col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Exemplo: alterar campos aninhados

Para alterar uma coluna em um campo aninhado, use:

ALTER TABLE table_name ALTER [COLUMN] col_name.nested_col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Por exemplo, se o esquema anterior à execução de ALTER TABLE boxes ALTER COLUMN colB.field2 FIRST for:

- root
| - colA
| - colB
| +-field1
| +-field2

o esquema após será:

- root
| - colA
| - colB
| +-field2
| +-field1

Substituir colunas

Use ALTER TABLE ... REPLACE COLUMNS para redefinir a lista de colunas completa de uma tabela, incluindo adicionar, remover, reordenar ou renomear colunas em uma única operação:

ALTER TABLE table_name REPLACE COLUMNS (col_name1 col_type1 [COMMENT col_comment1], ...)

Exemplo: Substituir campos aninhados

Por exemplo, ao executar a seguinte DDL:

ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT<field2:STRING, nested:STRING, field1:STRING>, colA STRING)

Se o esquema antes for:

- root
| - colA
| - colB
| +-field1
| +-field2

o esquema após será:

- root
| - colC
| - colB
| +-field2
| +-nested
| +-field1
| - colA

Renomear colunas

Para renomear colunas sem reescrever nenhum dos dados existentes das colunas, você deve habilitar o mapeamento de colunas para a tabela. Confira Renomear e remover colunas usando o mapeamento de colunas do Delta Lake.

Para renomear uma coluna:

ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name

Exemplo: renomear campos aninhados

Para renomear um campo aninhado:

ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field

Por exemplo, ao executar o seguinte comando:

ALTER TABLE boxes RENAME COLUMN colB.field1 TO field001

Se o esquema antes for:

- root
| - colA
| - colB
| +-field1
| +-field2

O esquema após será:

- root
| - colA
| - colB
| +-field001
| +-field2

Confira Renomear e remover colunas usando o mapeamento de colunas do Delta Lake.

Remover colunas

Para remover colunas como uma operação somente de metadados sem reescrever nenhum arquivo de dados, você precisa habilitar o mapeamento de colunas para a tabela. Confira Renomear e remover colunas usando o mapeamento de colunas do Delta Lake.

Note

A remoção de uma coluna dos metadados não exclui os dados subjacentes da coluna nos arquivos. Para limpar os dados da coluna descartada:

  • Use REORG TABLE para reescrever arquivos.
  • Em seguida, use VACUUM para excluir fisicamente os arquivos que contêm os dados de coluna descartados.

Para remover uma coluna:

ALTER TABLE table_name DROP COLUMN col_name

Para excluir/eliminar várias colunas:

ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)

Alterar o tipo ou o nome da coluna

Você pode alterar o tipo ou o nome de uma coluna ou soltar uma coluna reescrevendo a tabela. Para fazer isso, use a opção overwriteSchema.

O exemplo a seguir mostra a alteração de um tipo de coluna:

(spark.read.table(...)
  .withColumn("birthDate", col("birthDate").cast("date"))
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

O exemplo a seguir mostra a alteração de um nome de coluna:

(spark.read.table(...)
  .withColumnRenamed("dateOfBirth", "birthDate")
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

Habilitar a evolução do esquema

Use WITH SCHEMA EVOLUTION ou defina mergeSchema como true para fazer alterações de esquema com base no esquema dos dados que você deseja INSERT ou MERGE em uma tabela existente.

Habilite a evolução do esquema usando um dos seguintes métodos:

O Databricks recomenda habilitar a evolução do esquema para cada operação de gravação usando a WITH SCHEMA EVOLUTION sintaxe ou a opção mergeSchema em vez de definir uma configuração do Spark.

Quando você usa opções ou sintaxe para habilitar a evolução do esquema em uma operação de gravação, isso tem precedência sobre a configuração do Spark.

Habilitar a evolução de esquema para operações de escrita a fim de adicionar novas colunas

Quando a evolução do esquema está habilitada, as colunas presentes na consulta de origem, mas ausentes da tabela de destino, são adicionadas automaticamente como parte de uma transação de gravação. Consulte Habilitar a evolução do esquema.

Considere o seguinte:

  • As maiúsculas e minúsculas são preservadas no acréscimo de uma nova coluna.
  • Novas colunas são adicionadas ao final do esquema de tabelas.
  • Se as colunas adicionais estiverem em um struct, elas serão acrescentadas ao final do struct na tabela de destino.

INSERT com a evolução do schema usando SQL

Use a cláusula WITH SCHEMA EVOLUTION em instruções INSERT para permitir a evolução do esquema:

INSERT WITH SCHEMA EVOLUTION INTO target_table
SELECT * FROM source_table

Se a consulta em source_table retornar colunas que não existem na tabela de destino, essas colunas serão adicionadas automaticamente ao target_table esquema. As linhas existentes recebem NULL valores para as novas colunas.

INSERT com a evolução do esquema usando a API dataframe

O exemplo a seguir demonstra o uso da opção mergeSchema com uma operação de gravação em lote:

Python
(spark.read
  .table("source_table")
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("target_table")
)
Scala
spark.read
  .table("source_table")
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("target_table")

INSERT com a evolução do esquema com streaming estruturado

O exemplo a seguir demonstra o uso da opção mergeSchema com o Carregador Automático para Streaming Estruturado. Confira O que é o Carregador Automático?.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .trigger(availableNow=True)
  .toTable("table_name")
)

Evolução automática do esquema para mesclagem

Para MERGE, a evolução do esquema permite que você resolva incompatibilidades de esquema entre a tabela de destino e de origem. Ele lida com os dois casos a seguir:

  1. Existe uma coluna na tabela de origem, mas não na tabela de destino, e é especificada pelo nome em uma atribuição de ações de inserção ou atualização. Como alternativa, uma ação UPDATE SET * ou INSERT * está presente.

    Essa coluna será adicionada ao esquema de destino e seus valores serão preenchidos da coluna correspondente na origem.

    • Isso só se aplica quando o nome da coluna e a estrutura na fonte de mesclagem correspondem exatamente à atribuição de destino.

    • A nova coluna deve estar presente no esquema de origem. Atribuir a nova coluna na cláusula de ação não define essa coluna.

    Estes exemplos permitem a evolução do esquema:

    -- The column newcol is present in the source but not in the target. It will be added to the target.
UPDATE SET target.newcol = source.newcol

-- The field newfield doesn't exist in struct column somestruct of the target. It will be added to that struct column.
UPDATE SET target.somestruct.newfield = source.somestruct.newfield

-- The column newcol is present in the source but not in the target.
-- It will be added to the target.
UPDATE SET target.newcol = source.newcol + 1

-- Any columns and nested fields in the source that don't exist in target will be added to the target.
UPDATE SET *
INSERT *

    Esses exemplos não disparam a evolução do esquema se a coluna newcol não estiver presente no source esquema:

    UPDATE SET target.newcol = source.someothercol
UPDATE SET target.newcol = source.x + source.y
UPDATE SET target.newcol = source.output.newcol

  2. Existe uma coluna na tabela de destino, mas não na tabela de origem.

    O esquema de destino não é alterado. Estas colunas:

    • São deixados inalterados para UPDATE SET *.

    • Estão definidos para NULL de INSERT *.

    • Ainda poderá ser explicitamente modificado caso seja atribuído na cláusula de ação.

    Por exemplo:

    UPDATE SET *  -- The target columns that are not in the source are left unchanged.
INSERT *  -- The target columns that are not in the source are set to NULL.
UPDATE SET target.onlyintarget = 5  -- The target column is explicitly updated.
UPDATE SET target.onlyintarget = source.someothercol  -- The target column is explicitly updated from some other source column.

Você deve habilitar manualmente a evolução automática do esquema. Consulte Habilitar a evolução do esquema.

Note

No Databricks Runtime 11.3 LTS e abaixo, somente INSERT * ou UPDATE SET * ações podem ser usadas para evolução do esquema com mesclagem.

No Databricks Runtime 12.2 LTS e posteriores, colunas e campos de estrutura presentes na tabela de origem podem ser especificados pelo nome em ações INSERT ou UPDATE.

No Databricks Runtime 13.3 LTS e versões posteriores, você pode usar a evolução de esquema com structs aninhados em mapas, como map<int, struct<a: int, b: int>>.

MERGEcom a evolução do esquema usando SQL, Python e Scala

No Databricks Runtime 15.4 LTS ou posteriores, você pode especificar a evolução do esquema em um comando de mesclagem usando SQL ou APIs de tabela.

SQL
MERGE WITH SCHEMA EVOLUTION INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE
Python
from delta.tables import *

(targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)
Scala
import io.delta.tables._

targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

Operações de exemplo de MERGE com evolução de esquema

Aqui estão alguns exemplos dos efeitos da operação MERGE com e sem a evolução do esquema.

Columns Consulta (no SQL) Comportamento sem a evolução do esquema (padrão) Comportamento com a evolução do esquema
Colunas de destino: key, value
Colunas de origem: key, value, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 O esquema de tabela permanece inalterado; somente as colunas key, value são atualizadas/inseridas. O esquema da tabela é alterado para (key, value, new_value). Os registros existentes com correspondências são atualizados com value e new_value na origem. Novas linhas são inseridas com o esquema (key, value, new_value).
Colunas de destino: key, old_value
Colunas de origem: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 UPDATE e INSERT as ações geram um erro porque a coluna old_value de destino não está na origem. O esquema da tabela é alterado para (key, old_value, new_value). Os registros existentes com correspondências são atualizados com o new_value na origem deixando old_value inalterado. Novos registros são inseridos com os key, new_value e NULL especificados para o old_value.
Colunas de destino: key, old_value
Colunas de origem: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET new_value = s.new_value		 UPDATE gera um erro porque a coluna new_value não existe na tabela de destino. O esquema da tabela é alterado para (key, old_value, new_value). Os registros existentes com correspondências são atualizados com o new_value na origem deixando old_value inalterado e registros não correspondentes foram NULL inseridos para new_value. Confira a observação (1).
Colunas de destino: key, old_value
Colunas de origem: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN NOT MATCHED
THEN INSERT (key, new_value) VALUES (s.key, s.new_value)		 INSERT gera um erro porque a coluna new_value não existe na tabela de destino. O esquema da tabela é alterado para (key, old_value, new_value). Novos registros são inseridos com os key, new_value e NULL especificados para o old_value. Os registros existentes têm NULL inseridos para new_value deixando old_value inalterados. Confira a observação (1).

(1) Esse comportamento está disponível no Databricks Runtime 12.2 LTS e versões superiores; o Databricks Runtime 11.3 LTS e versões anteriores gera erro nessa condição.

Excluir colunas ao mesclar

No Databricks Runtime 12.2 LTS e posteriores, você pode usar cláusulas EXCEPT em condições de mesclagem para excluir explicitamente colunas. O comportamento da palavra-chave EXCEPT varia dependendo se a evolução do esquema está habilitada ou não.

Quando a evolução do esquema é desativada, a palavra-chave EXCEPT se aplica à lista de colunas da tabela de destino e permite excluir colunas de ações UPDATE ou INSERT. As colunas excluídas são definidas como null.

Com a evolução do esquema habilitada, a palavra-chave EXCEPT se aplica à lista de colunas na tabela de origem e permite a exclusão de colunas da evolução do esquema. Uma nova coluna na origem, não presente na tabela de destino, não será adicionada ao esquema de destino se estiver listada na EXCEPT cláusula. As colunas excluídas que já existem no destino são configuradas como null.

Exemplos do EXCLUDE com o MERGE

Os exemplos a seguir demonstram essa sintaxe:

Columns Consulta (no SQL) Comportamento sem a evolução do esquema (padrão) Comportamento com a evolução do esquema
Colunas de destino: id, title, last_updated
Colunas de origem: id, title, review, last_updated		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated)		 As linhas correspondentes são atualizadas definindo o campo last_updated como a data atual. Novas linhas são inseridas usando valores para id e title. O campo last_updated excluído é definido como null. O campo review é ignorado porque não está no destino. As linhas correspondentes são atualizadas definindo o campo last_updated como a data atual. O esquema é desenvolvido para adicionar o campo review. Novas linhas são inseridas usando todos os campos de origem, exceto last_updated, que é definida como null.
Colunas de destino: id, title, last_updated
Colunas de origem: id, title, review, internal_count		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated, internal_count)		 INSERT gera um erro porque a coluna internal_count não existe na tabela de destino. As linhas correspondentes são atualizadas definindo o campo last_updated como a data atual. O campo review é adicionado à tabela de destino, mas o campo internal_count é ignorado. Novas linhas inseridas têm last_updated definida como null.

Habilitar a evolução de esquemas com a configuração do Spark (legada)

Você pode definir a configuração do Spark spark.databricks.delta.schema.autoMerge.enabled para true para habilitar a evolução do esquema para todas as operações de gravação no SparkSession atual.

Python

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", True)

Scala

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", true)

SQL

SET spark.databricks.delta.schema.autoMerge.enabled=true

Note

O Databricks não recomenda essa abordagem para produção. Definir uma configuração em toda a sessão pode levar a alterações de esquema não intencionais em várias operações e torna mais difícil raciocinar sobre quais operações evoluem o esquema.

Em vez disso, habilite a evolução do esquema para cada operação de gravação:

Quando você usa opções ou sintaxe para habilitar a evolução do esquema em uma operação de gravação, isso tem precedência sobre a configuração do Spark.

Substituir esquema de tabela

Por padrão, substituir os dados em uma tabela não substitui o esquema. Ao substituir uma tabela usando mode("overwrite") sem replaceWhere, talvez você ainda queira substituir o esquema dos dados que estão sendo gravados.

Para substituir o esquema e o particionamento da tabela, defina a opção overwriteSchema como true:

df.write.option("overwriteSchema", "true")

Note

Você não pode especificar overwriteSchema como true ao usar a substituição de partição dinâmica. Consulte Substituições de partição dinâmica com partitionOverwriteMode (herdado).

  • Last updated on