如何在Delta Lake Merge操作中对匹配行同时执行更新与插入？

Delta Lake的Merge语法确实不支持whenMatchedInsert，因为Merge的设计逻辑是：匹配行仅执行更新/删除操作，未匹配行执行插入操作。要实现匹配时同时更新旧行+插入新行的需求（典型如缓慢变化维度SCD Type 2场景），可以利用Delta的事务特性保证操作原子性，分两步完成且依赖同一初始表状态。

核心思路 #

1. 先从目标表的当前版本中筛选出需要处理的匹配行（id匹配且state不同），构造要插入的新状态行。
2. 在同一个原子事务中，先执行Merge更新旧行的end_time，再插入新构造的状态行。事务会确保两步操作要么全部成功，要么全部失败，不会依赖中间修改后的表状态。

具体代码实现 #

假设目标表结构为id, state, start_time, end_time，新数据包含id, state, start_time字段：

import deltalake as dl
from deltalake import DeltaTable
import pyarrow as pa

# 加载目标表与新数据（示例数据）
target_table = DeltaTable("path/to/your/delta/table")
new_data = pa.Table.from_pylist([
    {"id": 1, "state": "active", "start_time": "2024-05-01"},
    {"id": 2, "state": "inactive", "start_time": "2024-05-01"}
])

# 1. 筛选需要更新和插入的记录
# 读取目标表当前版本数据
existing_records = target_table.to_pyarrow_table()
# 匹配id相同且state不同的记录
matched_ids = existing_records.join(new_data, keys="id", join_type="inner").filter(
    pa.compute.not_equal(pa.field("state_left"), pa.field("state_right"))
)["id"].unique()
# 构造要插入的新状态行：补充end_time字段（设为None表示当前有效状态）
new_data_with_end = new_data.append_column(
    "end_time", pa.array([None]*len(new_data), type=pa.string())
)
# 筛选出需要插入的行：匹配行对应的新数据 + 完全未匹配的新数据
inserts = new_data_with_end.filter(
    pa.field("id").isin(matched_ids) | 
    pa.compute.not_(pa.field("id").isin(existing_records["id"]))
)

# 2. 原子事务中执行更新+插入
with target_table.transaction() as txn:
    # 第一步：更新匹配到的旧行
    (
        txn.alias("existing")
        .merge(
            new_data.alias("new"),
            "existing.id = new.id"
        )
        .whenMatchedUpdate(
            condition="existing.state != new.state",
            set={"end_time": "new.start_time"}
        )
        .execute()
    )
    # 第二步：插入新状态行（避免重复插入）
    (
        txn.alias("existing")
        .merge(
            inserts.alias("new_rows"),
            """
            existing.id = new_rows.id 
            AND existing.state = new_rows.state 
            AND existing.start_time = new_rows.start_time
            """
        )
        .whenNotMatchedInsertAll()
        .execute()
    )

关键说明 #

• 事务的作用：事务内的所有操作基于事务启动时的表版本执行，避免了拆分操作导致的依赖中间状态问题。
• 避免重复插入：插入时使用Merge的whenNotMatchedInsertAll，确保不会插入已存在的相同状态行，保证数据唯一性。
• 性能优化：如果表数据量较大，可通过分区过滤、字段投影减少读取的数据量，提升效率。

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