问SCD-2在数据库中使用Delta

EN

//Base Table
 val employeeDf = Seq((1,"John","CT"),
                     (2,"Mathew","MA"),
                     (3,"Peter","CA"),
                     (4,"Joel","NY"))
                    .toDF("ID","NAME","ADDRESS")

val empBaseDf = employeeDf.withColumn("IS_ACTIVE",lit(1))
  .withColumn("EFFECTIVE_DATE",current_date())
  .withColumn("TERMINATION_DATE",lit(null).cast(StringType))      

empBaseDf.write.format("delta").mode("overwrite").saveAsTable("empBase")

// Batch Data
//Note: Here 1 record changed, 2 new record and 1 unchnaged.
val updateEmployeeDf = Seq( (1,"John","NH"),
                 (2,"Mathew","MA"),
                (5,"Adam","NJ"),
                (6,"Philip","CT")).toDF("ID","NAME","ADDRESS").createOrReplaceTempView("EmpUpdates")


val updatedEmp = updateEmployeeDf.withColumn("IS_ACTIVE",lit(1))
      .withColumn("EFFECTIVE_DATE",current_date())
      .withColumn("TERMINATION_DATE",lit(null).cast(StringType))      

updatedEmp.createOrReplaceTempView("empBatch")

import io.delta.tables._    
val empbaseTable: DeltaTable =  DeltaTable.forName("empBase")          
val empBatch = table("empBatch")

// Rows to INSERT new addresses of existing customers
val newAddressesToInsert = empBatch
  .as("batch")
  .join(empbaseTable.toDF.as("emp"), "ID")
  .where("batch.ADDRESS <> emp.ADDRESS").selectExpr("batch.*")

newAddressesToInsert.show()

val processRec = newAddressesToInsert
  .selectExpr("NULL as mergeKey", "*")
  .union(empBatch.selectExpr("ID as mergeKey", "*")  )                 
processRec.show()

empbaseTable
  .as("base")
  .merge(processRec.as("batch1"),"base.ID = mergeKey")
  .whenMatched("base.IS_ACTIVE = true AND base.address <> batch1.address")
  .updateExpr(Map(                                      
    "IS_ACTIVE" -> "false",
    "TERMINATION_DATE" -> "current_date()"))
  .whenNotMatched()  
  .insertExpr((Map("ID" -> "batch1.ID",
              "NAME" -> "batch1.NAME",
              "ADDRESS" -> "batch1.ADDRESS",             
              "IS_ACTIVE" -> "true",              
              "EFFECTIVE_DATE" -> "current_date()",
               "TERMINATION_DATE" -> "null" )))
  .execute()

//With multiple run of the above code duplicate records are getting inserted. I need to restrict the duplicate entry into the delta table.
ID  NAME    ADDRESS IS_ACTIVE   EFFECTIVE_DATE  TERMINATION_DATE
1   John    NH  1   2020-06-25  null
1   John    CT  0   2020-06-25  2020-06-25
1   John    NH  1   2020-06-25  null
2   Mathew  MA  1   2020-06-25  null
3   Peter   CA  1   2020-06-25  null
4   Joel    NY  1   2020-06-25  null
5   Adam    NJ  1   2020-06-25  null
6   Philip  CT  1   2020-06-25  null