Ampliação de tipo

Disponível em tabelas Delta Lake no Databricks Runtime 15.4 LTS e versões posteriores, o alargamento de tipos permite alterar os tipos de dados das colunas para um tipo mais amplo sem reescrever os ficheiros de dados.

Todas as tabelas gerenciadas pelo Unity Catalog usam Delta Lake por padrão. Consulte as tabelas geridas do Unity Catalog para Delta Lake e Apache Iceberg.

Alterações de tipo suportadas

Você pode ampliar os tipos de acordo com as seguintes regras:

Há suporte para alterações de tipo para colunas de nível superior e campos aninhados dentro de estruturas, mapas e matrizes.

Comportamento dos decimais

O Spark trunca a parte fracionária de um valor por defeito quando uma operação promove um tipo inteiro para um decimal ou um double e um processo de ingestão downstream escreve novamente o valor numa coluna inteira. Para mais detalhes sobre o comportamento das políticas de atribuição, consulte Store assignment.

Ao alterar qualquer tipo numérico para decimal, a precisão total deve ser igual ou superior à precisão inicial. Se aumentar também a escala, a precisão total deve aumentar em uma quantidade correspondente.

O objetivo mínimo para os tipos byte, shorte int é decimal(10,0). A meta mínima para long é decimal(20,0).

Se você quiser adicionar duas casas decimais a um campo com decimal(10,1), o destino mínimo é decimal(12,3).

Ativar expansão de tipo

Você pode habilitar a ampliação de tipo em uma tabela existente definindo a propriedade da tabela delta.enableTypeWidening como true.

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'true')

Você também pode habilitar a ampliação de tipos durante a criação da tabela.

  CREATE TABLE T(c1 INT) TBLPROPERTIES('delta.enableTypeWidening' = 'true')

Aplicar manualmente uma alteração de tipo

Use o comando ALTER COLUMN para alterar manualmente tipos:

ALTER TABLE <table_name> ALTER COLUMN <col_name> TYPE <new_type>

Esta operação atualiza o esquema de tabela sem reescrever os arquivos de dados subjacentes. Consulte ALTER TABLE para obter mais detalhes.

Amplie os tipos com evolução automática do esquema

Use evolução de esquema com alargamento de tipos para atualizar os tipos de dados nas tabelas alvo para corresponder ao tipo de dados recebidos.

Para usar a evolução do esquema para ampliar o tipo de dados de uma coluna durante a ingestão, você deve atender às seguintes condições:

• O comando write é executado com a evolução automática do esquema ativada.
• A tabela de destino tem a expansão de tipo ativada.
• O tipo de coluna de origem é maior do que o tipo de coluna de destino.
• O alargamento de tipo suporta a alteração de tipo.

Incompatibilidades de tipo que não atendem a todas essas condições seguem as regras normais de imposição de esquema. Consulte Aplicação do esquema.

Example

Os exemplos seguintes demonstram como o alargamento de tipos funciona com a evolução de esquemas.

Python

Crie uma tabela alvo com uma INT coluna e uma tabela de origem com uma BIGINT coluna:

spark.sql("CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true')")
spark.sql("CREATE TABLE source_table (id BIGINT, data STRING)")

Utilize saveAsTable() com evolução do esquema para alargar automaticamente a coluna INT para BIGINT durante uma anexação:

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

Utilizar MERGE INTO com evolução do esquema:

from delta.tables import DeltaTable

source_df = spark.table("source_table")
target_table = DeltaTable.forName(spark, "target_table")

(target_table.alias("target")
  .merge(source_df.alias("source"), "target.id = source.id")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .execute()
)

Scala

Crie uma tabela alvo com uma INT coluna e uma tabela de origem com uma BIGINT coluna:

spark.sql("CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true')")
spark.sql("CREATE TABLE source_table (id BIGINT, data STRING)")

Utilize saveAsTable() com evolução do esquema para alargar automaticamente a coluna INT para BIGINT durante uma anexação:

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

Utilizar MERGE INTO com evolução do esquema:

import io.delta.tables.DeltaTable

val sourceDf = spark.table("source_table")
val targetTable = DeltaTable.forName(spark, "target_table")

targetTable.alias("target")
  .merge(sourceDf.alias("source"), "target.id = source.id")
  .withSchemaEvolution()
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

SQL

Crie uma tabela alvo com uma INT coluna e uma tabela de origem com uma BIGINT coluna:

CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true');
CREATE TABLE source_table (id BIGINT, data STRING);

Utilize INSERT INTO com evolução do esquema para alargar automaticamente a coluna INT para BIGINT durante uma anexação:

INSERT WITH SCHEMA EVOLUTION INTO target_table SELECT * FROM source_table;

Utilizar MERGE INTO com evolução do esquema:

MERGE WITH SCHEMA EVOLUTION INTO target_table
USING source_table
ON target_table.id = source_table.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Carregador Automático

O Auto Loader suporta ampliação de tipos com evolução automática de esquemas. Quando usa o Auto Loader para ingerir dados numa tabela Delta Lake com alargamento de tipos e evolução de esquemas ativados, os tipos de coluna são automaticamente alargados para corresponder aos dados recebidos.

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

Veja "Expansão automática do tipo" com Auto Loader. Além disso, a tabela alvo deve ter o alargamento de tipos ativado. Consulte Ativar ampliação de tipo.

Desativar o recurso de tabela de alargamento de tipo

Você pode evitar o alargamento acidental de tipos em tabelas habilitadas definindo a propriedade como false:

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'false')

Esta definição impede futuras alterações de tipo à tabela, mas não remove a funcionalidade de alargamento de tipos nem desfaz alterações anteriores.

Se precisa remover completamente as funcionalidades da tabela de alargamento de tipo, pode usar o comando DROP FEATURE, conforme demonstrado no exemplo a seguir:

 ALTER TABLE <table-name> DROP FEATURE 'typeWidening' [TRUNCATE HISTORY]

Ao eliminar o alargamento de tipos, o Databricks reescreve todos os ficheiros de dados que não cumprem o esquema atual da tabela. Consulte Remover uma funcionalidade da tabela Delta Lake e fazer downgrade do protocolo da tabela.

Transmissão a partir de uma tabela Delta Lake

O suporte para expansão de tipos em Structured Streaming está disponível no Databricks Runtime 16.4 LTS e em versões superiores.

Ao transmitir a partir de uma tabela Delta Lake com alargamento de tipos ativado, pode configurar o alargamento automático de tipos para consultas de streaming ativando a evolução do esquema com a mergeSchema opção na tabela de destino. A tabela alvo deve ter a expansão de tipos ativada. Consulte Ativar ampliação de tipo.

Python

(spark.readStream
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", "/path/to/checkpointLocation")
  .option("mergeSchema", "true")
  .toTable("output_table")
)

Scala

spark.readStream
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", "/path/to/checkpointLocation")
  .option("mergeSchema", "true")
  .toTable("output_table")

Quando mergeSchema está ativado e a tabela de destino tem alargamento de tipos ativado:

• As alterações de tipo são aplicadas automaticamente à tabela a jusante sem necessidade de intervenção manual.
• Novas colunas são adicionadas automaticamente ao esquema da tabela a jusante.

Sem mergeSchema ativação, os valores são tratados de acordo com a spark.sql.storeAssignmentPolicy configuração, que por defeito reduz os valores para corresponder ao tipo de coluna alvo. Para mais informações sobre o comportamento das políticas de atribuição, consulte Atribuição de Armazenamento.

Tratar alterações de tipo num fluxo

Ao transmitir a partir de uma tabela Delta Lake, pode fornecer um local de rastreamento de esquema para acompanhar alterações não aditivas do esquema, incluindo alterações de tipo. Fornecer um local de rastreamento do esquema é necessário no Databricks Runtime 18.0 e anteriores, e é opcional no Databricks Runtime 18.1 e posteriores.

Não podes definir um schemaTrackingLocation usando SQL. Ver Funcionalidades não suportadas.

schemaTrackingLocation deve estar definido para um local no mesmo caminho que o seu checkpoint de streaming. Por exemplo:

Python

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

Scala

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

Depois de definir um local de rastreio do esquema, o fluxo atualiza o esquema que está a rastrear quando deteta uma alteração de tipo e, em seguida, para. Nessa altura, tem de lidar com a alteração de tipo, por exemplo ativando a expansão de tipos na tabela a jusante ou atualizando a consulta de fluxo.

Para retomar o processamento, defina a configuração spark.databricks.delta.streaming.allowSourceColumnTypeChange do Spark ou a DataFrame opção allowSourceColumnTypeChangedo leitor, como no seguinte exemplo:

Python

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  # alternatively to allow all future type changes for this stream:
  # .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

Scala

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  // alternatively to allow all future type changes for this stream:
  // .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

SQL

  -- To unblock for this particular stream just for this series of schema change(s):
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange.ckpt_<checkpoint_id> = "<delta_source_table_version>"
  -- To unblock for this particular stream:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "<delta_source_table_version>"
  -- To unblock for all streams:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "always"

Quando o fluxo termina, uma mensagem de erro mostra o ID <checkpoint_id> do ponto de controlo e a versão <delta_source_table_version>da tabela de origem do Delta Lake.

Para uma lista completa das opções de streaming em Delta Lake, veja Delta Lake.

Condutas de fluxo de lago

Pode ativar o alargamento de tipos para pipelines Lakeflow ao nível do pipeline ou para tabelas individuais. A ampliação de tipos permite que os tipos de coluna sejam ampliados automaticamente durante a execução do pipeline, sem necessidade de uma atualização completa das tabelas em fluxo contínuo. Alterações de tipo em vistas materializadas provocam sempre um recálculo completo, e quando uma alteração de tipo é aplicada a uma tabela de origem, as vistas materializadas que dependem dessa tabela requerem um recálculo completo para refletir os novos tipos.

Permitir o alargamento de tipo em toda a pipeline

Para ativar o alargamento de tipos para todas as tabelas de um pipeline, defina a configuração do pipeline pipelines.enableTypeWidening:

JSON

{
  "configuration": {
    "pipelines.enableTypeWidening": "true"
  }
}

YAML

configuration:
  pipelines.enableTypeWidening: 'true'

Permitir alargamento de tipos para tabelas específicas

Também pode ativar o alargamento de tipos para tabelas individuais definindo a propriedade delta.enableTypeWideningda tabela :

Python

import dlt

@dlt.table(
  table_properties={"delta.enableTypeWidening": "true"}
)
def my_table():
  return spark.readStream.table("source_table")

SQL

CREATE OR REFRESH STREAMING TABLE my_table
TBLPROPERTIES ('delta.enableTypeWidening' = 'true')
AS SELECT * FROM source_table

Compatibilidade com leitores a jusante

As tabelas com alargamento de tipo ativado só podem ser lidas no Databricks Runtime 15.4 LTS e superior. Se quiser que uma tabela com alargamento de tipos ativada no seu pipeline seja legível pelos leitores no Databricks Runtime 14.3 e inferiores, deve:

• Desligue o alargamento de tipos removendo a propriedade delta.enableTypeWidening/pipelines.enableTypeWidening ou definindo-a como falso, e acione uma atualização completa da tabela.
• Ativa o Modo de Compatibilidade na tua mesa.

OpenSharing

Partilhar uma tabela Delta Lake com alargamento de tipos ativado é suportado no OpenSharing Databricks-para-Databricks. O provedor e o destinatário devem estar no Databricks Runtime 16.1 ou superior.

Para ler o fluxo de dados de alterações de uma tabela Delta Lake com a expansão de tipos ativada com o OpenSharing, deve definir o formato de resposta como delta:

spark.read
  .format("deltaSharing")
  .option("responseFormat", "delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", "<start version>")
  .option("endingVersion", "<end version>")
  .load("<table>")

Não é suportado ler o feed de dados das alterações entre tipos. Em vez disso, você deve dividir a operação em duas leituras separadas, uma terminando na versão da tabela que contém a alteração de tipo e a outra começando na versão que contém a alteração de tipo.

Limitações

Compatibilidade com Apache Iceberg

O Apache Iceberg não suporta todas as alterações de tipo abrangidas pela expansão de tipos. Ver Evolução do Esquema do Iceberg.

As alterações de tipo não suportadas incluem as seguintes:

• byte, short, int, long para decimal ou double
• aumento da escala decimal
• date a timestampNTZ

Quando ativas as leituras Iceberg numa tabela Delta Lake, aplicar uma das alterações de tipo anteriores resulta num erro. Consulte Ler tabelas Delta Lake com clientes Iceberg.

Se aplicar uma destas alterações de tipo não suportadas a uma tabela Delta Lake, tem duas opções:

• Regenerar metadados do Iceberg: Use o seguinte comando para regenerar metadados do Iceberg sem a funcionalidade de tabela de alargamento de tipos:

  ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.universalFormat.config.icebergCompatVersion' = '<version>')
  

  Isto permite manter a compatibilidade com leituras Iceberg após aplicar alterações de tipo incompatíveis.

• Eliminar a funcionalidade de alargamento de tipos: Veja Desativar a funcionalidade de alargamento de tipos.

Funções dependentes do tipo

Algumas funções SQL retornam resultados que dependem do tipo de dado de entrada. Por exemplo, hash a função devolve valores hash diferentes para o mesmo valor lógico se o tipo de argumento for diferente: hash(1::INT) devolve um resultado diferente de hash(1::BIGINT).

Outras funções dependentes do tipo são: xxhash64, bit_get, bit_reverse, typeof.

Para resultados estáveis em consultas que utilizam estas funções, deve castar explicitamente valores para o tipo desejado:

Python

spark.read.table("table_name") \
  .selectExpr("hash(CAST(column_name AS BIGINT))")

Scala

spark.read.table("main.johan_lasperas.dlt_type_widening_bronze2")
  .selectExpr("hash(CAST(a AS BIGINT))")

SQL

-- Use explicit casting for stable hash values
SELECT hash(CAST(column_name AS BIGINT)) FROM table_name

Funcionalidades não suportadas

• Não podes definir uma localização de rastreio de esquema usando SQL ao fazer streaming de uma tabela Delta Lake com uma mudança de tipo.
• Não pode partilhar uma tabela com alargamento de tipos ativado para consumidores que não sejam Databricks usando OpenSharing.