upsert.scala

object DedupWriter extends Serializable {
  def upsertIntoDeduped(microBatchOutput: DataFrame, batchId: Long): Unit = {
    DeltaTable.forPath("/mnt/somebucket/ip_index_deduped_updates.delta").as("out")
      .merge(
        microBatchOutput.as("in"),
        // all columns match
      )
      .whenNotMatched.insertAll.execute
  }
}
spark
  .readStream
  .format("delta")
  .option("ignoreChanges", true)
  .load("/mnt/aggregates/prod/some_table_that_receives_upserts.delta")
  .writeStream
  .outputMode("append")
  .option("checkpointLocation", CHECKPOINT_PATH)
  .foreachBatch(DedupWriter.upsertIntoDeduped _)
  .start()

def upsert(microbatch: DataFrame, batchId: Long) {
    val batchDf = microbatch
    silverTable.as("m")
    .merge(
        batchDf.as("b"),
        "m.record_key_hash = b.record_key_hash"
    ).whenMatched("m.record_hash <> b.record_hash").updateExpr(Map(
        "m.name" -> "b.name"
    ))
    .whenNotMatched().insertAll()
    .execute()
}
spark.readStream.option("ignoreChanges", "true").format("delta").load("<bronze-table>").writeStream.format("delta").foreachBatch(upsert _).outputMode("update").option("checkpointLocation", "out/checkpoint/2").start("<silver-table>")

In Python there is no option of table