Carregar dados com fusão (merge), inserção (insert) e anexar (append)

Depois de transformar e limpar os dados, o passo final do pipeline é carregá-los nas tabelas de destino. O Azure Databricks fornece três estratégias principais de carregamento: anexar, sobrescrever e mesclar. Cada estratégia serve casos de uso diferentes, e escolher a certa depende de como os seus dados de origem se relacionam com os registos existentes.

Nesta unidade, aprende a acrescentar novas linhas com INSERT, substituir dados por operações de sobrescrever e sincronizar alterações usando MERGE para cenários upsert.

Adicionar dados com INSERT INTO

A operação de carregamento mais direta adiciona novas linhas a uma tabela existente sem modificar os dados atuais. Use INSERT INTO quando estiver a carregar novos lotes de dados que não coincidam com registos existentes — como registos diários de transações, novos fluxos de eventos ou extrações incrementais de dados.

INSERT INTO funciona tanto com SQL como com PySpark. Os exemplos seguintes mostram como adicionar linhas a uma tabela de vendas.

-- Append new records using VALUES
INSERT INTO sales.transactions (transaction_id, amount, transaction_date)
VALUES 
    ('TXN001', 150.00, '2026-01-15'),
    ('TXN002', 275.50, '2026-01-15');

-- Append from a staging table or query
INSERT INTO sales.transactions
SELECT * FROM staging.daily_transactions
WHERE transaction_date = current_date();

No PySpark, adiciona dados escrevendo um DataFrame com append modo:

# Append DataFrame to existing table
df_new_transactions.write.mode("append").saveAsTable("sales.transactions")

# Or use insertInto for existing tables
df_new_transactions.write.insertInto("sales.transactions")

Substituir dados por INSERT OVERWRITE

Por vezes, é preciso substituir dados existentes em vez de os acrescentar. INSERT OVERWRITE trunca a tabela ou partição de destino antes de inserir novas linhas. Esta abordagem funciona bem para reprocessar lotes falhados, atualizar tabelas de consulta ou reconstruir agregações.

-- Replace entire table contents
INSERT OVERWRITE sales.daily_summary
SELECT 
    region,
    SUM(amount) as total_sales,
    COUNT(*) as transaction_count
FROM sales.transactions
WHERE transaction_date = current_date()
GROUP BY region;

Para tabelas particionadas, pode sobrescrever partições específicas deixando outras intactas:

-- Replace only January 2026 partition
INSERT OVERWRITE sales.monthly_report
PARTITION (report_month = '2026-01')
SELECT region, product_category, SUM(amount) as revenue
FROM sales.transactions
WHERE transaction_date BETWEEN '2026-01-01' AND '2026-01-31'
GROUP BY region, product_category;

O Delta Lake também suporta sobrescrições seletivas com a REPLACE WHERE cláusula, que elimina linhas que correspondem a uma condição antes de inserir:

-- Replace transactions for a specific date range
INSERT INTO sales.transactions
REPLACE WHERE transaction_date >= '2026-01-01' AND transaction_date <= '2026-01-07'
SELECT * FROM staging.corrected_transactions;

Por defeito, cada linha que a consulta devolve deve corresponder ao predicado REPLACE WHERE. Se alguma linha ficar fora do predicado, a instrução falha e não escreve nada.

No PySpark, use overwrite o modo para substituição total da tabela e adicione a replaceWhere opção de substituir apenas as linhas que correspondam a uma condição:

# Overwrite entire table
df_refreshed.write.mode("overwrite").saveAsTable("sales.daily_summary")

# Replace only the January 2026 rows
(df_january.write
    .mode("overwrite")
    .option("replaceWhere", "report_month = '2026-01'")
    .saveAsTable("sales.monthly_report"))

Substituir dinamicamente os dados com REPLACE USING

REPLACE USING é a abordagem recomendada para sobrescrições dinâmicas de dados. Funciona em todos os tipos de computação — incluindo Databricks, SQL warehouses e computação serverless — e não requer configuração de sessões Spark. Ao contrário de partitionOverwriteMode, REPLACE USING suporta tabelas particionadas, tabelas não particionadas e tabelas com clustering líquido.

-- Replace only the partitions touched by the incoming data
INSERT INTO sales.monthly_report
  REPLACE USING (report_month)
  SELECT region, product_category, SUM(amount) AS revenue, report_month
  FROM staging.corrected_transactions
  GROUP BY region, product_category, report_month;

A cláusula REPLACE USING elimina atomicamente as linhas em sales.monthly_report que correspondem às linhas de entrada em report_month, e em seguida insere as novas linhas. As linhas que não correspondem às colunas especificadas ficam intocadas. No Databricks Runtime 16.3–17.1, as USING colunas devem ser o conjunto completo de colunas de partição da tabela (comportamento legado). No Databricks Runtime 17.2 e superiores, quaisquer colunas de qualquer tipo de tabela — incluindo tabelas não particionadas e tabelas com cluster líquido — são suportadas. A cláusula SQL REPLACE USING exige Databricks Runtime 16.3 ou superior, e a opção Python e Scala replaceUsing requer Databricks Runtime 18.2 ou superior. Como REPLACE USING usa igualdade normal, as linhas com valores NULL nas colunas USING não coincidem nem são removidas da tabela de destino.

Quando precisar de um emparelhamento mais flexível — como substituir linhas com base numa condição personalizada ou tratar valores NULL como iguais — use REPLACE ON em vez disso:

-- Replace rows matching a user-defined condition across source and target
INSERT INTO sales.transactions AS target
  REPLACE ON (target.transaction_id = source.transaction_id)
  (SELECT * FROM staging.corrected_transactions) AS source;

REPLACE ON requer o Databricks Runtime 17.1 ou superior para SQL e o Databricks Runtime 18.2 ou superior para a opção replaceOn em Python e Scala. Use REPLACE WHERE (já referido) quando precisar de substituição baseada em condição sem referência na tabela de origem.

Fusão de dados para operações upsert

Cadeias de processamento de dados do mundo real muitas vezes precisam atualizar registos existentes enquanto inserem novos. A instrução MERGE lida com este padrão — comumente chamado de "upsert" — ao corresponder as linhas de origem às linhas de destino e aplicar diferentes ações consoante a existência de uma correspondência.

O MERGE é essencial para dimensões de mudança lenta, sincroniza dados dos sistemas de origem e processa fluxos de captura de dados de alteração (CDC).

MERGE INTO customers AS target
USING customer_updates AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET 
        target.email = source.email,
        target.phone = source.phone,
        target.last_updated = current_timestamp()
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email, phone, created_date, last_updated)
    VALUES (source.customer_id, source.name, source.email, source.phone, 
            current_timestamp(), current_timestamp());

Esta consulta atualiza o email e o telefone dos clientes existentes, inserindo registos de clientes totalmente novos. A cláusula ON define a chave correspondente — tipicamente uma chave primária ou identificador empresarial.

Fusão com lógica condicional

Adicione condições às cláusulas WHEN para lidar com cenários mais complexos. Por exemplo, pode atualizar apenas registos que realmente mudaram:

MERGE INTO products AS target
USING product_feed AS source
ON target.sku = source.sku
WHEN MATCHED AND source.price <> target.price THEN
    UPDATE SET target.price = source.price, target.updated_at = current_timestamp()
WHEN MATCHED AND source.discontinued = true THEN
    DELETE
WHEN NOT MATCHED THEN
    INSERT *;

A INSERT * sintaxe insere todas as colunas da fonte quando correspondem ao esquema de destino.

Fusão com o PySpark

Para controlo programático, use a API Python do Delta Lake:

from delta.tables import DeltaTable

target_table = DeltaTable.forName(spark, "customers")

target_table.alias("target").merge(
    source_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdate(set={
    "email": "source.email",
    "phone": "source.phone",
    "last_updated": "current_timestamp()"
}).whenNotMatchedInsert(values={
    "customer_id": "source.customer_id",
    "name": "source.name",
    "email": "source.email",
    "phone": "source.phone",
    "created_date": "current_timestamp()",
    "last_updated": "current_timestamp()"
}).execute()

Para casos mais simples em que os esquemas de origem e destino coincidem, use os métodos de conveniência:

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

Escolha a estratégia de carregamento certa

A sua escolha de operação de carregamento depende da relação entre os dados de origem e de destino:

Compreender estes padrões de carregamento prepara-o para construir pipelines de dados fiáveis que mantêm a integridade dos dados enquanto processam eficientemente tanto registos novos como alterados.