Ladda in data med sammanfoga, infoga och addera

Efter att du har omvandlat och rensat data är det sista steget i din pipeline att läsa in dem i måltabeller. Azure Databricks innehåller tre grundläggande inläsningsstrategier: lägga till, skriva över och slå samman. Varje strategi hanterar olika användningsfall, och att välja rätt beror på hur dina källdata relaterar till befintliga poster.

I den här lektionen får du lära dig hur du lägger till nya rader med INSERT, ersätter data med överskrivningsåtgärder och synkroniserar ändringar med MERGE för scenarier med upsert.

Lägga till data med INSERT INTO

Den enklaste inläsningsåtgärden lägger till nya rader i en befintlig tabell utan att ändra aktuella data. Använd INSERT INTO när du läser in nya batchar med data som inte överlappar befintliga poster, till exempel dagliga transaktionsloggar, nya händelseströmmar eller inkrementella dataextrakt.

INSERT INTO fungerar med både SQL och PySpark. I följande exempel visas hur du lägger till rader i en försäljningstabell.

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

I PySpark lägger du till data genom att skriva en DataFrame med append läge:

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

Ersätt data med INSERT OVERWRITE

Ibland måste du ersätta befintliga data i stället för att lägga till dem. INSERT OVERWRITE trunkerar måltabellen eller partitionen innan nya rader infogas. Den här metoden fungerar bra för att bearbeta misslyckade batchar, uppdatera uppslagstabeller eller återskapa aggregeringar.

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

För partitionerade tabeller kan du skriva över specifika partitioner samtidigt som andra lämnas intakta:

-- 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 stöder också selektiva överskrivningar med REPLACE WHERE -satsen, som tar bort rader som matchar ett villkor innan du infogar:

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

Som standardinställning måste varje rad som frågan returnerar matcha predikatet REPLACE WHERE. Om någon rad hamnar utanför predikatet misslyckas instruktionen och skriver ingenting.

I PySpark använder du overwrite läget för fullständig tabellersättning och lägger till replaceWhere alternativet för att ersätta endast de rader som matchar ett villkor:

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

Ersätt data dynamiskt med REPLACE USING

REPLACE USING är den rekommenderade metoden för dynamiska dataöverskrivningar. Det fungerar för alla beräkningstyper, inklusive Databricks SQL-lager och serverlös beräkning, och kräver inte Spark-sessionskonfiguration. Till skillnad från partitionOverwriteModestöder REPLACE USING , partitionerade tabeller, opartitionerade tabeller och tabeller med flytande klustring.

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

REPLACE USING Klausulen tar atomiskt bort rader i sales.monthly_report som matchar inkommande rader på report_month, och infogar sedan de nya raderna. Rader som inte matchar de angivna kolumnerna lämnas orörda. I Databricks Runtime 16.3–17.1 måste kolumnerna USING vara tabellens fullständiga uppsättning partitionskolumner (äldre beteende). I Databricks Runtime 17.2 och senare stöds alla kolumner i valfri tabelltyp, inklusive opartitionerade tabeller och tabeller med flytande klustring. SQL-klausulen REPLACE USING kräver Databricks Runtime 16.3 eller senare, och alternativet Python och Scala replaceUsing kräver Databricks Runtime 18.2 eller senare. Eftersom REPLACE USING använder vanlig likhet matchar inte rader med NULL-värden i kolumnerna USING och tas inte bort från måltabellen.

När du behöver mer flexibel matchning, till exempel att ersätta rader baserat på ett anpassat villkor eller behandla NULL-värden som lika, använder du REPLACE ON i stället:

-- 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 ONkräver Databricks Runtime 17.1 och senare för SQL och Databricks Runtime 18.2 och senare för alternativet Python och Scala replaceOn. Använd REPLACE WHERE (beskrivs tidigare) när du behöver villkorsbaserad ersättning utan källtabellreferens.

Sammanfoga data för upsert-åtgärder

Verkliga datapipelines behöver ofta uppdatera befintliga poster vid infogning av nya. MERGE-instruktionen hanterar det här mönstret , som ofta kallas "upsert", genom att matcha källrader till målrader och tillämpa olika åtgärder baserat på om en matchning finns.

MERGE är viktigt för långsamt föränderliga dimensioner, synkronisering av data från källsystem och bearbetning av 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());

Den här frågan/kommandot uppdaterar e-post och telefon för befintliga kunder, samtidigt som helt nya kundposter infogas. ON-satsen definierar matchande nyckel – vanligtvis en primärnyckel eller affärsidentifierare.

Sammanfoga med villkorsstyrd logik

Lägg till villkor i WHEN-satser för att hantera mer komplexa scenarier. Du kanske till exempel bara uppdaterar poster som faktiskt har ändrats:

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

Syntaxen INSERT * infogar alla kolumner från källan när de matchar målschemat.

Slå samman med PySpark

För programmatisk kontroll använder du Delta Lake Python-API:et:

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

Använd bekvämlighetsmetoderna för enklare fall där käll- och målscheman matchar:

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

Välj rätt inläsningsstrategi

Valet av inläsningsåtgärd beror på relationen mellan käll- och måldata:

Om du förstår dessa inläsningsmönster kan du skapa tillförlitliga datapipelines som upprätthåller dataintegriteten samtidigt som du effektivt bearbetar både nya och ändrade poster.