Merge işlemi kullanılarak Delta Lake tablosuna güncelleme ve ekleme yapma

SQL işlemini kullanarak kaynak tablo, görünüm veya DataFrame'den hedef Delta Lake tablosuna MERGE veri ekleyebilirsiniz. Delta Lake, eklemeleri, güncellemeleri ve silmeleri MERGE ile destekler ve ileri düzeydeki kullanım senaryolarını kolaylaştırmak için SQL standartlarının ötesinde genişletilmiş bir söz dizimini destekler.

Diyelim ki, people10mupdates adlı bir kaynak tablonuz veya /tmp/delta/people-10m-updates konumunda bir kaynak yolunuz var ve bunlar, people10m adlı bir hedef tablo veya /tmp/delta/people-10m konumunda bir hedef yol için yeni veriler içeriyor. Bu yeni kayıtlardan bazıları hedef verilerde zaten mevcut olabilir. Yeni verileri birleştirmek için, kişinin id zaten mevcut olduğu satırları güncellemek ve eşleşen id bulunmayan yeni satırları birleştirmek istiyorsunuz. Aşağıdaki sorguyu çalıştırabilirsiniz:

SQL

MERGE INTO people10m
USING people10mupdates
ON people10m.id = people10mupdates.id
WHEN MATCHED THEN
  UPDATE SET
    id = people10mupdates.id,
    firstName = people10mupdates.firstName,
    middleName = people10mupdates.middleName,
    lastName = people10mupdates.lastName,
    gender = people10mupdates.gender,
    birthDate = people10mupdates.birthDate,
    ssn = people10mupdates.ssn,
    salary = people10mupdates.salary
WHEN NOT MATCHED
  THEN INSERT (
    id,
    firstName,
    middleName,
    lastName,
    gender,
    birthDate,
    ssn,
    salary
  )
  VALUES (
    people10mupdates.id,
    people10mupdates.firstName,
    people10mupdates.middleName,
    people10mupdates.lastName,
    people10mupdates.gender,
    people10mupdates.birthDate,
    people10mupdates.ssn,
    people10mupdates.salary
  )

Piton

from delta.tables import *

deltaTablePeople = DeltaTable.forName(spark, "people10m")
deltaTablePeopleUpdates = DeltaTable.forName(spark, "people10mupdates")

dfUpdates = deltaTablePeopleUpdates.toDF()

deltaTablePeople.alias('people') \
  .merge(
    dfUpdates.alias('updates'),
    'people.id = updates.id'
  ) \
  .whenMatchedUpdate(set =
    {
      "id": "updates.id",
      "firstName": "updates.firstName",
      "middleName": "updates.middleName",
      "lastName": "updates.lastName",
      "gender": "updates.gender",
      "birthDate": "updates.birthDate",
      "ssn": "updates.ssn",
      "salary": "updates.salary"
    }
  ) \
  .whenNotMatchedInsert(values =
    {
      "id": "updates.id",
      "firstName": "updates.firstName",
      "middleName": "updates.middleName",
      "lastName": "updates.lastName",
      "gender": "updates.gender",
      "birthDate": "updates.birthDate",
      "ssn": "updates.ssn",
      "salary": "updates.salary"
    }
  ) \
  .execute()

Scala programlama dili

import io.delta.tables._
import org.apache.spark.sql.functions._

val deltaTablePeople = DeltaTable.forName(spark, "people10m")
val deltaTablePeopleUpdates = DeltaTable.forName(spark, "people10mupdates")
val dfUpdates = deltaTablePeopleUpdates.toDF()

deltaTablePeople
  .as("people")
  .merge(
    dfUpdates.as("updates"),
    "people.id = updates.id")
  .whenMatched
  .updateExpr(
    Map(
      "id" -> "updates.id",
      "firstName" -> "updates.firstName",
      "middleName" -> "updates.middleName",
      "lastName" -> "updates.lastName",
      "gender" -> "updates.gender",
      "birthDate" -> "updates.birthDate",
      "ssn" -> "updates.ssn",
      "salary" -> "updates.salary"
    ))
  .whenNotMatched
  .insertExpr(
    Map(
      "id" -> "updates.id",
      "firstName" -> "updates.firstName",
      "middleName" -> "updates.middleName",
      "lastName" -> "updates.lastName",
      "gender" -> "updates.gender",
      "birthDate" -> "updates.birthDate",
      "ssn" -> "updates.ssn",
      "salary" -> "updates.salary"
    ))
  .execute()

Önemli

Kaynak tablodan yalnızca tek bir satır hedef tablodaki belirli bir satırla eşleşebilir. Databricks Runtime 16.0 ve üzeri sürümlerde, MERGE, WHEN MATCHED ve ON maddelerinde belirtilen koşulları, yinelenen eşleşmeleri belirlemek için değerlendirir. Databricks Runtime 15.4 LTS ve altında, MERGE işlemler belirtilen koşulları içeren ON yan tümcesini yalnızca dikkate alır.

Scala ve Python söz dizimi ayrıntıları için Delta Lake API belgelerine bakın. SQL söz dizimi ayrıntıları için bkz. MERGE INTO

Birleştirmeyi kullanarak eşleşmeyen tüm satırları değiştirme

Databricks SQL ve Databricks Runtime 12.2 LTS ve üzeri sürümlerde, kaynak tabloda karşılık gelen kayıtları olmayan kayıtları hedef tablodan WHEN NOT MATCHED BY SOURCE tümcesini kullanarak UPDATE veya DELETE yapabilirsiniz. Databricks, hedef tablonun tam olarak yeniden yazılmasını önlemek için isteğe bağlı bir koşullu yan tümce eklenmesini önerir.

Aşağıdaki kod örneği, silme işlemleri için bunu kullanmanın temel söz dizimini, hedef tablonun üzerine kaynak tablonun içeriğiyle birlikte yazılmasını ve hedef tablodaki eşleşmeyen kayıtların silinmesini gösterir. Kaynak güncelleştirmelerin ve silmelerin zamana bağlı olduğu tablolar için daha ölçeklenebilir bir desen için bkz. Delta Lake tablosunu kaynakla artımlı olarak eşitleme.

Piton

(targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)

Scala programlama dili

targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

SQL

MERGE INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE

Aşağıdaki örnek yan tümcesine WHEN NOT MATCHED BY SOURCE koşullar ekler ve eşleşmeyen hedef satırlarda güncelleştirilecek değerleri belirtir.

Piton

(targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatchedUpdate(
    set = {"target.lastSeen": "source.timestamp"}
  )
  .whenNotMatchedInsert(
    values = {
      "target.key": "source.key",
      "target.lastSeen": "source.timestamp",
      "target.status": "'active'"
    }
  )
  .whenNotMatchedBySourceUpdate(
    condition="target.lastSeen >= (current_date() - INTERVAL '5' DAY)",
    set = {"target.status": "'inactive'"}
  )
  .execute()
)

Scala programlama dili

targetDF
  .merge(sourceDF, "source.key = target.key")
  .whenMatched()
  .updateExpr(Map("target.lastSeen" -> "source.timestamp"))
  .whenNotMatched()
  .insertExpr(Map(
    "target.key" -> "source.key",
    "target.lastSeen" -> "source.timestamp",
    "target.status" -> "'active'",
    )
  )
  .whenNotMatchedBySource("target.lastSeen >= (current_date() - INTERVAL '5' DAY)")
  .updateExpr(Map("target.status" -> "'inactive'"))
  .execute()

SQL

MERGE INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET target.lastSeen = source.timestamp
WHEN NOT MATCHED THEN
  INSERT (key, lastSeen, status) VALUES (source.key,  source.timestamp, 'active')
WHEN NOT MATCHED BY SOURCE AND target.lastSeen >= (current_date() - INTERVAL '5' DAY) THEN
  UPDATE SET target.status = 'inactive'

Birleştirme işlemi semantiği

Programlı işlem semantiğinin merge ayrıntılı bir açıklaması aşağıdadır.

  • herhangi bir sayıda whenMatched ve whenNotMatched yan tümcesi olabilir.

  • whenMatched yan tümceleri, bir kaynak satırı eşleşme koşuluna göre bir hedef tablo satırıyla eşleştiğinde yürütülür. Bu yan tümceler aşağıdaki semantiklere sahiptir.

    • whenMatched yan tümcelerinin en fazla bir update ve bir delete eylemi olabilir. update içindeki merge eylemi, eşleşen hedef satırın belirtilen sütunlarını yalnızca (updateişleme benzer şekilde) güncelleştirir. Eylem, delete eşleşen satırı siler.

    • Her whenMatched yan tümce isteğe bağlı bir koşula sahip olabilir. Bu yan tümce koşulu varsa, update veya delete eylemi yalnızca yan tümce koşulu true olduğunda eşleşen herhangi bir kaynak hedef satır çifti için yürütülür.

    • Birden çok whenMatched yan tümce varsa, bunlar belirtildikleri sırayla değerlendirilir. Sonuncusu dışındaki tüm whenMatched yan tümceler koşul içermelidir.

    • Birleştirme koşuluyla whenMatched eşleşen bir kaynak ve hedef satır çifti için koşulların hiçbiri true olarak değerlendirilmezse, hedef satır değişmeden bırakılır.

    • Hedef Delta Lake tablosunun tüm sütunlarını kaynak veri kümesinin ilgili sütunlarıyla güncelleştirmek için kullanın whenMatched(...).updateAll(). Bu, şuna eşdeğerdir:

      whenMatched(...).updateExpr(Map("col1" -> "source.col1", "col2" -> "source.col2", ...))

      hedef Delta Lake tablosunun tüm sütunları için. Bu nedenle, bu eylem kaynak tablonun hedef tablodakilerle aynı sütunlara sahip olduğunu varsayar, aksi takdirde sorgu bir çözümleme hatası oluşturur.

      Not

      Otomatik şema evrimi etkinleştirildiğinde bu davranış değişir. Ayrıntılar için bkz . otomatik şema evrimi .

  • whenNotMatched yan tümceleri, bir kaynak satırı eşleşme koşuluna göre herhangi bir hedef satırla uyuşmadığında yürütülür. Bu yan tümceler aşağıdaki semantiklere sahiptir.

    • whenNotMatched yan tümceleri yalnızca insert eylemini içerebilir. Yeni satır, belirtilen sütuna ve karşılık gelen ifadelere göre oluşturulur. Hedef tablodaki tüm sütunları belirtmeniz gerekmez. Belirtilmemiş hedef sütunlar için NULL eklenir.

    • Her whenNotMatched yan tümce isteğe bağlı bir koşula sahip olabilir. Şarta bağlı tümce varsa, yalnızca bu şart satır için doğruysa bir kaynak satır eklenir. Aksi takdirde, kaynak sütun yoksayılır.

    • Birden çok whenNotMatched yan tümce varsa, bunlar belirtildikleri sırayla değerlendirilir. Sonuncusu dışındaki tüm whenNotMatched yan tümceler koşul içermelidir.

    • Hedef Delta Lake tablosunun tüm sütunlarını kaynak veri kümesinin ilgili sütunlarıyla birlikte eklemek için kullanın whenNotMatched(...).insertAll(). Bu, şuna eşdeğerdir:

      whenNotMatched(...).insertExpr(Map("col1" -> "source.col1", "col2" -> "source.col2", ...))

      hedef Delta Lake tablosunun tüm sütunları için. Bu nedenle, bu eylem kaynak tablonun hedef tablodakilerle aynı sütunlara sahip olduğunu varsayar, aksi takdirde sorgu bir çözümleme hatası oluşturur.

      Not

      Otomatik şema evrimi etkinleştirildiğinde bu davranış değişir. Ayrıntılar için bkz . otomatik şema evrimi .

  • whenNotMatchedBySource yan tümceleri, bir hedef satır birleştirme koşuluna göre hiçbir kaynak satırla eşleşmediğinde yürütülür. Bu yan tümceler aşağıdaki semantiklere sahiptir.

    • whenNotMatchedBySourcetümceler delete belirtebilir ve update eylemleri.
    • Her whenNotMatchedBySource yan tümce isteğe bağlı bir koşula sahip olabilir. Yan tümce koşulu varsa, hedef satır, yalnızca bu koşul söz konusu satır için doğru olduğunda değiştirilir. Aksi takdirde, hedef satır değişmeden bırakılır.
    • Birden çok whenNotMatchedBySource yan tümce varsa, bunlar belirtildikleri sırayla değerlendirilir. Sonuncusu dışındaki tüm whenNotMatchedBySource yan tümceler koşul içermelidir.
    • Tanım gereği, whenNotMatchedBySource yan tümcelerde sütun değerlerini alacak bir kaynak satır bulunmadığından, kaynak sütunlara başvurulamaz. Değiştirilecek her sütun için ya bir sabit belirtebilir ya da SET target.deleted_count = target.deleted_count + 1gibi bir eylemi hedef sütunda gerçekleştirebilirsiniz.

Önemli

  • merge Kaynak veri kümesinin birden çok satırı eşleşirse ve birleştirme işlemi hedef Delta Lake tablosunun aynı satırlarını güncelleştirmeye çalışırsa işlem başarısız olabilir. Birleştirmenin SQL semantiğine göre, eşleşen hedef satırı güncelleştirmek için hangi kaynak satırın kullanılacağı belirsiz olduğundan bu tür bir güncelleştirme işlemi belirsizdir. Birden çok eşleşme olasılığını ortadan kaldırmak için kaynak tabloyu önceden işleyebilirsiniz.
  • Bir SQL GÖRÜNÜMÜne SQL MERGE işlemi uygulayabilmeniz için görünümün CREATE VIEW viewName AS SELECT * FROM deltaTable olarak tanımlanmış olması gerekir.

Delta Lake tablolarına yazarken yinelenen verileri kaldırma

Günlükleri bir tabloya ekleyerek bir Delta Lake tablosunda toplamak, yaygın bir ETL kullanım senaryosudur. Ancak, genellikle kaynaklar yinelenen günlük kayıtları oluşturabilir ve bunları yönetmek için sonraki adımlarda yinelenenlerin kaldırılması gerekir. ile merge, yinelenen kayıtları eklemekten kaçınabilirsiniz.

SQL

MERGE INTO logs
USING newDedupedLogs
ON logs.uniqueId = newDedupedLogs.uniqueId
WHEN NOT MATCHED
  THEN INSERT *

Piton

deltaTable.alias("logs").merge(
    newDedupedLogs.alias("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId") \
  .whenNotMatchedInsertAll() \
  .execute()

Scala programlama dili

deltaTable
  .as("logs")
  .merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId")
  .whenNotMatched()
  .insertAll()
  .execute()

Java

deltaTable
  .as("logs")
  .merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId")
  .whenNotMatched()
  .insertAll()
  .execute();

Not

Yeni günlükleri içeren veri kümesinin içindeki yinelemeleri gidermesi gerekir. Birleştirmenin SQL semantiğiyle, yeni verileri tablodaki mevcut verilerle eşleştirir ve yinelenenleri kaldırır, ancak yeni veri kümesinde yinelenen veriler varsa eklenir. Bu nedenle, tabloya birleştirmeden önce yeni verilerdeki tekrar edenleri kaldırın.

Yalnızca birkaç gün boyunca yinelenen kayıtlar alabileceğinizi biliyorsanız, tabloyu tarihe göre bölümleyerek ve ardından eşleşecek hedef tablonun tarih aralığını belirterek sorgunuzu daha da iyileştirebilirsiniz.

SQL

MERGE INTO logs
USING newDedupedLogs
ON logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS
WHEN NOT MATCHED AND newDedupedLogs.date > current_date() - INTERVAL 7 DAYS
  THEN INSERT *

Piton

deltaTable.alias("logs").merge(
    newDedupedLogs.alias("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS") \
  .whenNotMatchedInsertAll("newDedupedLogs.date > current_date() - INTERVAL 7 DAYS") \
  .execute()

Scala programlama dili

deltaTable.as("logs").merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS")
  .whenNotMatched("newDedupedLogs.date > current_date() - INTERVAL 7 DAYS")
  .insertAll()
  .execute()

Java

deltaTable.as("logs").merge(
    newDedupedLogs.as("newDedupedLogs"),
    "logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYS")
  .whenNotMatched("newDedupedLogs.date > current_date() - INTERVAL 7 DAYS")
  .insertAll()
  .execute();

Bu, tablonun tamamında değil yalnızca son 7 günlükte yinelenenleri ararken önceki komuttan daha verimlidir. Ayrıca, kayıtların sürekli tekrarlananlarını kaldırmak için bu salt eklemeli birleştirmeyi Structured Streaming ile kullanabilirsiniz.

  • Akış sorgusunda, foreachBatch içindeki birleştirme işlemini kullanarak yinelenen kayıtları kaldırıp her türlü akış verisini bir Delta Lake tablosuna sürekli olarak yazabilirsiniz. Daha fazla bilgi için aşağıdaki akış örneğine bakın.
  • Başka bir akış sorgusunda, bu Delta Lake tablosundan tekilleştirilmiş verileri sürekli okuyabilirsiniz. Bu mümkündür çünkü yalnızca ekleme birleştirme işlemi Delta Lake tablosuna yalnızca yeni veriler ekler.

Delta Lake ile yavaş değişen veriler (SCD) ve veri yakalamayı (CDC) değiştirme

Lakeflow işlem hatları, SCD Tür 1 ve Tür 2'yi izlemek ve uygulamak için yerel desteğe sahiptir. CDC akışlarını işlerken sırası bozuk kayıtların doğru şekilde işlendiğinden emin olmak için Lakeflow işlem ardışık düzenleriyle birlikte AUTO CDC ... INTO kullanın. Bkz. AUTO CDC API'leri: Veri akışları ile değişiklik verilerini yakalamayı basitleştirme.

Delta Lake tablosunu kaynak ile artımlı olarak eşitle

WHEN NOT MATCHED BY SOURCE kullanarak Databricks SQL ve Databricks Runtime 12.2 LTS ve sonrasında, bir tablonun bir bölümünü atomik olarak silmek ve değiştirmek için keyfi koşullar oluşturabilirsiniz. Bu, özellikle kayıtların ilk veri girdisi sonrasında birkaç gün boyunca değişebileceği veya silinebileceği, ancak sonunda son duruma geçebileceği bir kaynak tablonuz olduğunda yararlı olabilir.

Aşağıdaki sorguda, kaynaktan 5 günlük kayıtları seçmek, hedefteki eşleşen kayıtları güncelleştirmek, kaynaktan hedefe yeni kayıtlar eklemek ve hedefte son 5 günün eşleşmeyen tüm kayıtlarını silmek için bu desenin kullanılması gösterilmektedir.

MERGE INTO target AS t
USING (SELECT * FROM source WHERE created_at >= (current_date() - INTERVAL '5' DAY)) AS s
ON t.key = s.key
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE AND created_at >= (current_date() - INTERVAL '5' DAY) THEN DELETE

Kaynak ve hedef tablolarda aynı boole filtresini sağlayarak, silme işlemleri de dahil olmak üzere değişiklikleri kaynağınızdan hedef tablolara dinamik olarak yayabilirsiniz.

Not

Bu desen herhangi bir koşullu yan tümce olmadan kullanılabilse de, bu durum hedef tablonun tamamen yeniden yazılmasıyla sonuçlanabilir ve bu da pahalı olabilir.

  • Last updated on