Gegevens laden met samenvoegen, invoegen en toevoegen

Nadat u gegevens hebt getransformeerd en opgeschoond, bestaat de laatste stap in uw pijplijn uit het laden van deze gegevens in de doeltabellen. Azure Databricks biedt drie kernstrategieën voor laden: toevoegen, overschrijven en samenvoegen. Elke strategie dient verschillende use cases en het kiezen van de juiste strategie is afhankelijk van hoe uw brongegevens zich verhouden tot bestaande records.

In deze les leert u hoe u nieuwe rijen toevoegt met INSERT, gegevens vervangt door overschrijfbewerkingen en wijzigingen synchroniseert met behulp van MERGE voor upsert-scenario's.

Gegevens toevoegen met INSERT INTO

Met de eenvoudigste laadbewerking worden nieuwe rijen aan een bestaande tabel toegevoegd zonder de huidige gegevens te wijzigen. Gebruik INSERT INTO wanneer u nieuwe batches met gegevens laadt die niet overlappen met bestaande records, zoals dagelijkse transactielogboeken, nieuwe gebeurtenisstromen of incrementele gegevensextracten.

INSERT INTO werkt zowel met SQL als PySpark. In de volgende voorbeelden ziet u hoe u rijen toevoegt aan een verkooptabel.

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

In PySpark voegt u gegevens toe door een DataFrame te schrijven met append de modus:

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

Gegevens vervangen door INSERT OVERWRITE

Soms moet u bestaande gegevens vervangen in plaats van eraan toe te voegen. INSERT OVERWRITE truncheert de doeltabel of doelpartitie voordat u nieuwe rijen invoegt. Deze aanpak werkt goed voor het opnieuw verwerken van mislukte batches, het vernieuwen van opzoektabellen of het herbouwen van aggregaties.

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

Voor gepartitioneerde tabellen kunt u specifieke partities overschrijven terwijl anderen intact blijven:

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

Delta Lake biedt ook ondersteuning voor selectieve overschrijfbewerkingen met de REPLACE WHERE component, waarmee rijen worden verwijderd die overeenkomen met een voorwaarde voordat u het invoegt:

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

Standaard moet elke rij die door de query wordt geretourneerd, overeenkomen met het REPLACE WHERE predicaat. Als een willekeurige rij buiten het predicaat valt, mislukt de instructie en wordt niets geschreven.

Gebruik in PySpark de modus overwrite voor volledige tabelvervanging en voeg de optie replaceWhere toe om alleen de rijen te vervangen die overeenkomen met een voorwaarde:

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

Gegevens dynamisch vervangen door REPLACE USING

REPLACE USING is de aanbevolen methode voor het overschrijven van dynamische gegevens. Het werkt voor alle rekentypen, waaronder Databricks SQL-warehouses en serverloze berekeningen, en vereist geen Spark-sessieconfiguratie. In tegenstelling tot partitionOverwriteMode, REPLACE USING ondersteunt gepartitioneerde tabellen, niet-gepartitioneerde tabellen en tabellen met vloeibare clustering.

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

De REPLACE USING clausule verwijdert atomisch rijen in sales.monthly_report die overeenkomen met binnenkomende rijen op report_month, en voegt vervolgens de nieuwe rijen in. Rijen die niet overeenkomen met de opgegeven kolommen worden ongemoeid gelaten. In Databricks Runtime 16.3-17.1 moeten de kolommen de USING volledige set partitiekolommen van de tabel zijn (verouderd gedrag). In Databricks Runtime 17.2 en hoger worden kolommen op elk tabeltype, inclusief niet-gepartitioneerde tabellen en tabellen met liquide clustering, ondersteund. Voor de SQL-component REPLACE USING is Databricks Runtime 16.3 of hoger vereist. Voor de optie Python en Scala replaceUsing is Databricks Runtime 18.2 of hoger vereist. Omdat REPLACE USING gewone gelijkheid gebruikt, komen rijen met NULL-waarden in de USING-kolommen niet overeen en worden ze niet verwijderd uit de doeltabel.

Wanneer u flexibelere overeenkomsten nodig hebt, zoals het vervangen van rijen op basis van een aangepaste voorwaarde of het behandelen van NULL-waarden als gelijk, gebruikt REPLACE ON u in plaats daarvan:

-- 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 ONvereist Databricks Runtime 17.1 en hoger voor SQL en Databricks Runtime 18.2 en hoger voor de optie Python en Scala replaceOn. Gebruik REPLACE WHERE (eerder behandeld) wanneer u op voorwaarde gebaseerde vervanging nodig hebt zonder verwijzing naar een brontabel.

Gegevens samenvoegen voor upsert-bewerkingen

Echte gegevenspijplijnen moeten vaak bestaande records bijwerken tijdens het invoegen van nieuwe records. De merge-instructie verwerkt dit patroon, ook wel 'upsert' genoemd, door bronrijen te koppelen aan doelrijen en verschillende acties toe te passen op basis van of er een overeenkomst bestaat.

MERGE is essentieel voor langzaam veranderende dimensies, het synchroniseren van gegevens uit bronsystemen en het verwerken van CDC-feeds (Change Data Capture).

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

Met deze query worden e-mail en telefoon bijgewerkt voor bestaande klanten tijdens het invoegen van volledig nieuwe klantrecords. De ON-component definieert de overeenkomende sleutel, meestal een primaire sleutel of bedrijfs-id.

Samenvoegen met voorwaardelijke logica

Voorwaarden toevoegen aan WHEN-componenten om complexere scenario's af te handelen. U kunt bijvoorbeeld alleen records bijwerken die daadwerkelijk zijn gewijzigd:

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 *;

Met INSERT * de syntaxis worden alle kolommen uit de bron ingevoegd wanneer ze overeenkomen met het doelschema.

Samenvoegen met PySpark

Gebruik voor programmatisch beheer de Delta Lake Python-API:

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

Gebruik de handige methoden voor eenvoudigere gevallen waarin bron- en doelschema's overeenkomen:

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

De juiste laadstrategie kiezen

Uw keuze voor het laden is afhankelijk van de relatie tussen bron- en doelgegevens:

Als u deze laadpatronen begrijpt, kunt u betrouwbare gegevenspijplijnen bouwen die de gegevensintegriteit behouden terwijl nieuwe en gewijzigde records efficiënt worden verwerkt.