deltalake的merge場景是怎么樣的

MERGE INTO logsUSING newDedupedLogsON logs.uniqueId = newDedupedLogs.uniqueIdWHEN NOT MATCHED  THEN INSERT *

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

MERGE INTO logsUSING newDedupedLogsON logs.uniqueId = newDedupedLogs.uniqueId AND logs.date > current_date() - INTERVAL 7 DAYSWHEN NOT MATCHED AND newDedupedLogs.date > current_date() - INTERVAL 7 DAYS  THEN INSERT *

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

val customersTable: DeltaTable = ...   // table with schema (customerId, address, current, effectiveDate, endDate)
val updatesDF: DataFrame = ...          // DataFrame with schema (customerId, address, effectiveDate)
// Rows to INSERT new addresses of existing customersval newAddressesToInsert = updatesDF  .as("updates")  .join(customersTable.toDF.as("customers"), "customerid")  .where("customers.current = true AND updates.address <> customers.address")
// Stage the update by unioning two sets of rows// 1. Rows that will be inserted in the whenNotMatched clause// 2. Rows that will either update the current addresses of existing customers or insert the new addresses of new customersval stagedUpdates = newAddressesToInsert  .selectExpr("NULL as mergeKey", "updates.*")   // Rows for 1.  .union(    updatesDF.selectExpr("updates.customerId as mergeKey", "*")  // Rows for 2.  )
// Apply SCD Type 2 operation using mergecustomersTable  .as("customers")  .merge(    stagedUpdates.as("staged_updates"),    "customers.customerId = mergeKey")  .whenMatched("customers.current = true AND customers.address <> staged_updates.address")  .updateExpr(Map(                                      // Set current to false and endDate to source's effective date.    "current" -> "false",    "endDate" -> "staged_updates.effectiveDate"))  .whenNotMatched()  .insertExpr(Map(    "customerid" -> "staged_updates.customerId",    "address" -> "staged_updates.address",    "current" -> "true",    "effectiveDate" -> "staged_updates.effectiveDate",  // Set current to true along with the new address and its effective date.    "endDate" -> "null"))  .execute()

val deltaTable: DeltaTable = ... // DeltaTable with schema (key, value)// DataFrame with changes having following columns// - key: key of the change// - time: time of change for ordering between changes (can replaced by other ordering id)// - newValue: updated or inserted value if key was not deleted// - deleted: true if the key was deleted, false if the key was inserted or updatedval changesDF: DataFrame = ...// Find the latest change for each key based on the timestamp// Note: For nested structs, max on struct is computed as// max on first struct field, if equal fall back to second fields, and so on.val latestChangeForEachKey = changesDF  .selectExpr("key", "struct(time, newValue, deleted) as otherCols" )  .groupBy("key")  .agg(max("otherCols").as("latest"))  .selectExpr("key", "latest.*")deltaTable.as("t")  .merge(    latestChangeForEachKey.as("s"),    "s.key = t.key")  .whenMatched("s.deleted = true")  .delete()  .whenMatched()  .updateExpr(Map("key" -> "s.key", "value" -> "s.newValue"))  .whenNotMatched("s.deleted = false")  .insertExpr(Map("key" -> "s.key", "value" -> "s.newValue"))  .execute()

import io.delta.tables.*
val deltaTable = DeltaTable.forPath(spark, "/data/aggregates")
// Function to upsert microBatchOutputDF into Delta table using mergedef upsertToDelta(microBatchOutputDF: DataFrame, batchId: Long) {  deltaTable.as("t")    .merge(      microBatchOutputDF.as("s"),      "s.key = t.key")    .whenMatched().updateAll()    .whenNotMatched().insertAll()    .execute()}
// Write the output of a streaming aggregation query into Delta tablestreamingAggregatesDF.writeStream  .format("delta")  .foreachBatch(upsertToDelta _)  .outputMode("update")  .start()