如何在Hive中基于主键合并增量数据行？

针对你这种每日增量更新、按主键合并、字段多达100-150个的场景，手动写每个字段的合并逻辑显然不现实，这里提供两种高效的实现方案，核心思路都是避免硬编码大量字段，利用Hive元数据动态生成处理逻辑。

前提准备 #

首先确保你的增量数据带有加载时间标记（比如load_dt字段，值为数据加载的日期），如果没有，每次加载增量时可以通过current_date()给新增/变更记录打上时间戳——这是判断数据版本新旧的关键。假设：

• 生产表：prod_table，主键id，包含100+业务字段
• 每日增量表：delta_table，结构和生产表完全一致，包含当日新增/变更记录

方案一：窗口函数+动态SQL生成（适合生成全量快照） #

这个方案会把历史生产数据和当日增量数据合并，通过窗口函数取每个id下每个字段的最新非空值，最后生成全量的最新视图。

步骤1：动态生成字段处理逻辑 #

因为字段太多，我们直接从Hive元数据中读取字段列表，自动生成每个字段的first_value逻辑（ignore nulls会跳过空值，取时间最晚的非空值）。

可以用Shell脚本实现自动生成SQL：

# 1. 从元数据获取所有业务字段（排除主键id和加载时间load_dt）
cols=$(hive -e "
select concat(
    'first_value(', column_name, ' ignore nulls) over (partition by id order by load_dt desc) as ', column_name
) 
from information_schema.columns 
where table_name = 'prod_table' 
  and column_name not in ('id', 'load_dt')
order by ordinal_position;" | grep -v "WARN")

# 2. 拼接完整SQL
full_sql="
with all_combined_data as (
    -- 合并生产表历史数据和增量表当日数据
    select id, col1, col2, ..., load_dt from prod_table
    union all
    select id, col1, col2, ..., load_dt from delta_table
)
-- 去重+取每个id的最新字段值
select distinct
    id,
    $cols
from all_combined_data;"

# 3. 执行SQL，输出结果或写入新表
hive -e "$full_sql"

原理说明 #

first_value(col ignore nulls) over (partition by id order by load_dt desc)会对每个id分组，按加载时间倒序排列，取第一个非空的字段值——也就是该字段的最新有效数据。最后用distinct去重，得到每个id唯一的最新完整记录。

方案二：Hive MERGE语句（适合直接更新生产表，Hive 2.2+支持） #

如果你的Hive版本在2.2及以上，可以用MERGE语句直接将增量数据合并到生产表中，这样生产表始终保持最新状态，不需要每次生成全量快照。

步骤1：动态生成MERGE的更新/插入逻辑 #

同样利用元数据自动生成所有字段的更新规则（只更新增量中非空的字段，保留生产表原有的非空值）：

# 1. 生成更新字段的逻辑：仅当增量字段非空时覆盖生产表字段
update_clauses=$(hive -e "
select concat(
    column_name, ' = case when s.', column_name, ' is not null then s.', column_name, ' else t.', column_name, ' end'
)
from information_schema.columns 
where table_name = 'prod_table' 
  and column_name not in ('id', 'load_dt')
order by ordinal_position;" | grep -v "WARN")

# 2. 生成插入字段列表和值列表
insert_cols=$(hive -e "
select column_name 
from information_schema.columns 
where table_name = 'prod_table'
order by ordinal_position;" | grep -v "WARN" | tr '\n' ',' | sed 's/,$//')

insert_vals=$(hive -e "
select concat('s.', column_name) 
from information_schema.columns 
where table_name = 'prod_table'
order by ordinal_position;" | grep -v "WARN" | tr '\n' ',' | sed 's/,$//')

# 3. 拼接MERGE SQL
merge_sql="
MERGE INTO prod_table t
USING delta_table s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET
    $update_clauses
WHEN NOT MATCHED THEN INSERT ($insert_cols)
VALUES ($insert_vals);"

# 4. 执行MERGE，更新生产表
hive -e "$merge_sql"

原理说明 #

• WHEN MATCHED：当增量表和生产表的id匹配时，只更新增量表中非空的字段，避免覆盖生产表已有的有效数据。
• WHEN NOT MATCHED：当增量表中的id在生产表中不存在时，直接插入这条新记录。

注意事项 #

1. 时间标记的必要性：无论用哪种方案，必须有load_dt（或类似的批次标识）来确定数据的先后顺序，否则无法判断哪个是最新的变更。
2. 字段一致性：确保增量表和生产表的字段结构完全一致（字段名、类型相同），否则动态生成的SQL会出错。
3. 特殊字段处理：如果字段名包含特殊字符（比如空格、下划线以外的符号），需要在动态生成SQL时添加转义处理。

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