Spark（Scala）Inner Join实现Type1 SCD功能未达预期，求帮助

Hey，我刚接触Spark时也踩过这个Type1 CDC的坑！只用Inner Join确实达不到预期，因为它只能处理源和目标匹配上的记录，完全漏掉了需要插入的新数据。咱们一步步来解决这个问题：

先明确你的需求和数据场景 #

你的目标是Type1 CDC：

• 源和目标匹配到的记录：用源的最新值覆盖目标
• 源有但目标没有的记录：直接插入到目标

先把你的数据补全更清晰：

SRC_DF（当日源数据） #

TGT_DF（现有目标数据） #

关联键是(Account_nbr, Location_cd)对吧？

正确的Type1实现步骤（Scala代码） #

我们需要拆分更新和插入两个逻辑，再合并结果：

1. 定义关联键 #

先把要用来匹配的字段存起来，方便后续复用：

val joinKeys = Seq("Account_nbr", "Location_cd")

2. 处理匹配记录的更新逻辑 #

用Inner Join找到两边都有的记录，然后直接用源数据的字段覆盖目标的（因为Type1就是直接覆盖旧值）：

// 匹配到的记录：用SRC的最新值替换TGT的旧数据
val updatedRecords = SRC_DF.join(TGT_DF, joinKeys, "inner")
  .select(
    SRC_DF("Account_nbr"),
    SRC_DF("Location_cd"),
    SRC_DF("State_IN").as("State_IN"),
    SRC_DF("REF_IN").as("REF_IN")
  )

3. 处理新增记录的插入逻辑 #

用Left Anti Join筛选出源有但目标没有的记录——这个Join类型专门用来找左边有右边没有的数据，正好是我们需要插入的新数据：

// 未匹配到的记录：直接取SRC的数据作为插入项
val newRecordsToInsert = SRC_DF.join(TGT_DF, joinKeys, "left_anti")

4. 合并更新和插入的结果 #

用unionByName把两个DataFrame合并（用unionByName是为了保证字段顺序不影响结果）：

val finalTargetDF = updatedRecords.unionByName(newRecordsToInsert)

5. 验证最终结果 #

运行后你会得到预期的Type1 CDC结果：

为什么之前只用Inner Join不行？ #

Inner Join只会返回两边都存在的交集数据，完全忽略了源中新增的、目标没有的记录。而Type1 CDC需要同时处理“更新现有”和“插入新增”两种场景，所以必须结合Left Anti Join和Union操作才能完整实现。

内容的提问来源于stack exchange，提问作者Sidd