如何在PySpark中实现Hive表的Upsert操作？

需求对应逻辑拆解 #

• 当目标表中存在patientnumber与待插入数据一致，且casenumber也与待插入数据一致的记录时，执行原样更新（字段值与源数据保持一致）
• 当目标表中无对应patientnumber的记录，或存在patientnumber但casenumber不匹配时，插入新行

方案1：使用MERGE INTO（Hive 2.0+ 支持，推荐） #

该方案需要目标表为ACID事务表，需满足以下前提：

• Hive开启ACID支持（配置hive.support.concurrency=true、hive.enforce.bucketing=true等）
• 表存储格式为ORC
• 表属性设置transactional=true

执行代码 #

-- 若目标表未创建，先创建ACID事务表
CREATE TABLE IF NOT EXISTS patient_cases (
    patientnumber STRING,
    casenumber STRING,
    other_fields STRING -- 替换为你的实际字段
)
STORED AS ORC
TBLPROPERTIES ('transactional'='true');

-- 执行Upsert操作
MERGE INTO patient_cases target
USING (
    -- 替换为你的待插入数据来源，比如临时表或查询语句
    SELECT patientnumber, casenumber, other_fields FROM your_source_data
) source
-- 匹配条件：patientnumber和casenumber都相同
ON target.patientnumber = source.patientnumber AND target.casenumber = source.casenumber
WHEN MATCHED THEN
    -- 原样更新所有字段（与源数据一致）
    UPDATE SET
        patientnumber = source.patientnumber,
        casenumber = source.casenumber,
        other_fields = source.other_fields
WHEN NOT MATCHED THEN
    -- 插入新行
    INSERT (patientnumber, casenumber, other_fields)
    VALUES (source.patientnumber, source.casenumber, source.other_fields);

逻辑说明 #

• ON子句精准匹配需要更新的记录（patientnumber+casenumber双匹配）
• 匹配成功时执行UPDATE，确保记录与源数据完全一致
• 匹配失败的场景（无对应patientnumber，或patientnumber存在但casenumber不同）统一执行INSERT

方案2：临时表方式（兼容旧版Hive，无ACID要求） #

如果你的Hive版本不支持ACID事务，可以通过临时表+数据合并的方式实现Upsert：

执行代码 #

-- 1. 创建临时表存储待插入数据
CREATE TABLE IF NOT EXISTS staging_patient_cases LIKE patient_cases;
INSERT INTO staging_patient_cases
-- 插入你的待处理数据
SELECT * FROM your_input_data;

-- 2. 筛选目标表中不需要更新的记录（即与临时表patientnumber+casenumber不匹配的旧记录）
CREATE TABLE IF NOT EXISTS temp_retained_records AS
SELECT target.*
FROM patient_cases target
LEFT JOIN staging_patient_cases source
ON target.patientnumber = source.patientnumber AND target.casenumber = source.casenumber
WHERE source.patientnumber IS NULL;

-- 3. 合并数据并覆盖写入目标表
INSERT OVERWRITE TABLE patient_cases
SELECT * FROM temp_retained_records
UNION ALL
SELECT * FROM staging_patient_cases;

-- 4. 清理临时表（可选）
DROP TABLE staging_patient_cases;
DROP TABLE temp_retained_records;

逻辑说明 #

• 先将待插入数据存入临时表，再筛选出目标表中无需更新的旧记录
• 通过UNION ALL合并旧记录和新数据，再覆盖写入目标表，实现“保留无需更新的旧数据+插入/更新新数据”的效果
• 注意：INSERT OVERWRITE会清空原表数据，需确保临时表数据逻辑正确

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