Delta表执行Update操作导致中间DataFrame为空问题及解决

基于Spark执行Delta表数据转换处理时出现异常，初始处理逻辑如下：

val filteringRecordsToExpire = collectAllActiveRecords.join(collectingSrcSysIdsToExpire, Seq("trans_id"), "leftsemi") 

// filteringRecordsToExpire 存储所有需要标记为无效的ID
val expiredList = filteringRecordsToExpire.select("trans_id").distinct().collect()

expiredList.foreach(v => expireRecords(v(0).toString)) // 此处逐条更新对应记录

初始设计逻辑为：将上述步骤中已标记过期的同批次ID替换为新值后，重新写入同一张Delta表。其中collectingSrcSysIdsToExpire是存储所有待修改后插入表的ID的DataFrame，但实际执行时，跑完过期标记操作（本质是对表中同ID记录执行Update更新）后，该DataFrame直接变为空。先后尝试持久化该DataFrame、注册为临时表等方案，均未解决问题。

最终通过调整操作顺序、使用MERGE INTO语法解决问题：
由于trans_id为唯一事务ID，使用MERGE INTO不会产生重复数据问题，执行效率也更高。原先「先执行Update标记过期、再执行Insert写入新值」的逻辑无法正常运行，调整操作顺序后流程可正常跑通：

• 先从源数据提取去重的trans_id，将更新后的新值先INSERT写入目标表
• 再通过MERGE INTO匹配对应trans_id的旧记录，将旧条目标记为过期

核心实现代码如下：

val filteringRecordsToExpire = delta.join(collectingSrcSysIdsToExpire, Seq("trans_id"), "leftsemi")
.distinct()

// 先写入更新后的新数据
collectingSrcSysIdsToExpire.select(TargetTable.schema.map(f => col(f.name)): _*).write.insertInto(Table)

// 再通过MERGE标记旧数据过期
val sqlUpdateQry =
      s""" MERGE INTO TargetTable as tgtTable
           USING expireSrsIds as source
           ON tgtTable.trans_id = source.trans_id
           AND <其他业务匹配条件>
           WHEN MATCHED
           THEN UPDATE SET
           <旧记录过期字段更新逻辑>
"""

经实测验证：先执行INSERT写入新数据、再执行UPDATE/MERGE标记旧数据过期的顺序可稳定运行，先UPDATE后INSERT的逻辑会触发中间DataFrame为空的异常。

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