Carregar dados com mesclagem, inserção e acréscimo

Depois de transformar e limpar dados, a etapa final do pipeline é carregá-los em tabelas de destino. O Azure Databricks fornece três estratégias de carregamento principais: acrescentar, substituir e mesclar. Cada estratégia atende a diferentes casos de uso e escolher a certa depende de como os dados de origem se relacionam com os registros existentes.

Nesta unidade, você aprenderá a acrescentar novas linhas com INSERT, substituir dados com operações de substituição e sincronizar alterações usando MERGE para cenários de execução de upsert.

Acrescentar dados com INSERT INTO

A operação de carregamento mais simples adiciona novas linhas a uma tabela existente sem modificar os dados atuais. Use INSERT INTO quando estiver carregando novos lotes de dados que não se sobrepõem aos registros existentes, como logs de transações diárias, novos fluxos de eventos ou extrações de dados incrementais.

INSERT INTO funciona com SQL e PySpark. Os exemplos a seguir 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, você acrescenta dados escrevendo um DataFrame no modo append.

# 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 usando o comando INSERT OVERWRITE

Às vezes, você precisa substituir os dados existentes em vez de adicioná-los. INSERT OVERWRITE trunca a tabela ou partição de destino antes de inserir novas linhas. Essa abordagem funciona bem para reprocessar lotes com falha, atualizar tabelas de pesquisa ou recriar 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, você pode substituir 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 dá suporte a substituições seletivas com a cláusula REPLACE WHERE, que exclui linhas que correspondem a uma condição antes de inseri-las.

-- 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 padrão, cada linha retornada pela consulta deve corresponder ao REPLACE WHERE predicado. Se qualquer linha ficar fora do predicado, a instrução falhará e não gravará nada.

No PySpark, use overwrite o modo para substituição completa da tabela e adicione a opção replaceWhere para substituir apenas as linhas que correspondem 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 dados dinamicamente com REPLACE USING

REPLACE USING é a abordagem recomendada para substituições de dados dinâmicos. Ele funciona em todos os tipos de computação, incluindo databricks SQL warehouses e computação sem servidor, e não requer configuração de sessão do Spark. Ao contrário de partitionOverwriteMode, REPLACE USING dá suporte a tabelas particionadas, tabelas não particionadas e tabelas com agrupamento dinâmico.

-- 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 exclui atomicamente as linhas em sales.monthly_report que correspondem às linhas de entrada de report_month e, em seguida, insere as novas linhas. As linhas que não correspondem às colunas especificadas são deixadas intocadas. No Databricks Runtime 16.3–17.1, as USING colunas devem ser o conjunto completo de colunas de partição da tabela (comportamento herdado). No Databricks Runtime 17.2 e versões posteriores, há suporte para qualquer coluna em qualquer tipo de tabela, incluindo tabelas não particionadas e com clustering líquido. A cláusula SQL REPLACE USING requer o Databricks Runtime 16.3 ou superior e a opção Python e Scala replaceUsing requer o Databricks Runtime 18.2 ou superior. Como REPLACE USING usa igualdade regular, as linhas com valores NULL nas colunas USING não correspondem e não são removidas da tabela de destino.

Quando você precisar de correspondência mais flexível, como substituir linhas com base em uma 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 ONrequer o Databricks Runtime 17.1 e superior para SQL e Databricks Runtime 18.2 e superior para a opção Python e ScalareplaceOn. Use REPLACE WHERE (coberto anteriormente) quando precisar de substituição baseada em condição sem uma referência de tabela de origem.

Mesclar dados para operações de upsert

Os pipelines de dados do mundo real geralmente precisam atualizar os registros existentes ao inserir novos. A instrução MERGE lida com esse padrão, geralmente chamado de "upsert", combinando linhas de origem com linhas de destino e aplicando ações diferentes com base em se existe uma correspondência.

MERGE é essencial para dimensões de alteração lenta, sincronizando dados de sistemas de origem e processando feeds CDC (captura de dados de alteração).

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

Essa consulta atualiza o email e o telefone para clientes existentes, inserindo registros de clientes totalmente novos. A cláusula ON define a chave correspondente, normalmente uma chave primária ou um identificador de negócios.

Mesclar com lógica condicional

Adicione condições às cláusulas WHEN para lidar com cenários mais complexos. Por exemplo, você só pode atualizar registros que realmente foram alterados:

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 origem quando correspondem ao esquema de destino.

Mesclar com o PySpark

Para controle programático, use a API do 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 de destino correspondem, 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 correta

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

Entender esses padrões de carregamento prepara você para criar pipelines de dados confiáveis que mantêm a integridade dos dados, processando com eficiência registros novos e alterados.