MR引擎下Hive窗口函数HQL优化求助：大分区引发Reducer运行异常

嗨，我来帮你分析下这个HQL的优化思路，你的问题核心其实就是数据倾斜——少数upp_sys_id+ssn_seq_all的分区记录数远超其他（10万+ vs 1千以内），导致对应的Reducer要处理巨量数据，而窗口函数的计算又依赖全分区数据在同一个Reducer内完成，直接拖慢了整个任务。下面给你几个实用的优化方案：

一、预处理聚合，分离窗口函数的计算压力 #

你的原SQL里同时在窗口里计算min(bhv_tm)和row_number()，可以把min(bhv_tm)提前预聚合，减少Reducer的实时计算量：

with pre_min_bhv as (
    -- 提前计算每个分区的最小bhv_tm，Map端就能做部分聚合，Reducer压力更小
    select upp_sys_id, ssn_seq_all, min(bhv_tm) as min_bhv_tm
    from test.VT_seq_all
    group by upp_sys_id, ssn_seq_all
)
select 
    concat(t.upp_sys_id,'#',p.min_bhv_tm,'#',t.ssn_seq_all) as ssn_id,
    t.evt_drt,
    row_number() over (partition by t.upp_sys_id,t.ssn_seq_all order by t.bhv_tm) as ssn_seq,
    t.`dw_dat_dt`, t.`msg_id`, t.`evt_nm`, t.`upp_sys_id`, t.`bhv_tm`
from test.VT_seq_all t
join pre_min_bhv p 
    on t.upp_sys_id = p.upp_sys_id and t.ssn_seq_all = p.ssn_seq_all;

这样修改后，窗口函数只需要处理row_number()的排序计数，min(bhv_tm)用预聚合的结果，能大幅降低大分区Reducer的计算负载。

二、拆分超大分区，分散Reducer负载 #

如果预聚合后大分区的row_number()计算还是慢，可以把超大分区拆分成多个子分片，让多个Reducer并行处理：

-- 先区分正常分区和超大分区，分别处理后合并结果
select 
    concat(upp_sys_id,'#',min_bhv_tm,'#',ssn_seq_all) as ssn_id,
    evt_drt,
    ssn_seq,
    `dw_dat_dt`, `msg_id`, `evt_nm`, `upp_sys_id`, `bhv_tm`
from (
    -- 处理正常分区（记录数<1万）
    select 
        upp_sys_id, ssn_seq_all, evt_drt, `dw_dat_dt`, `msg_id`, `evt_nm`, `bhv_tm`,
        min(bhv_tm) over (partition by upp_sys_id,ssn_seq_all) as min_bhv_tm,
        row_number() over (partition by upp_sys_id,ssn_seq_all order by bhv_tm) as ssn_seq
    from (
        select *, count(1) over(partition by upp_sys_id,ssn_seq_all) as part_cnt
        from test.VT_seq_all
    ) t
    where part_cnt < 10000

    union all

    -- 处理超大分区，用哈希拆分成分片，并行计算row_number
    select 
        upp_sys_id, ssn_seq_all, evt_drt, `dw_dat_dt`, `msg_id`, `evt_nm`, `bhv_tm`,
        min_bhv_tm,
        row_number() over (partition by upp_sys_id,ssn_seq_all order by bhv_tm) as ssn_seq
    from (
        select 
            t.*, p.min_bhv_tm,
            mod(hash(bhv_tm), 10) as split_key -- 拆成10个分片，可根据实际调整数量
        from (
            select *, count(1) over(partition by upp_sys_id,ssn_seq_all) as part_cnt
            from test.VT_seq_all
        ) t
        join pre_min_bhv p 
            on t.upp_sys_id = p.upp_sys_id and t.ssn_seq_all = p.ssn_seq_all
        where part_cnt >= 10000
    ) t
) final;

这里通过mod(hash(bhv_tm), 10)把大分区拆成10个子分片，让多个Reducer同时处理，最后再计算全局的row_number，避免单个Reducer扛下所有大分区数据。

三、调整Hive参数，优化Reducer资源和执行效率 #

针对MR引擎的特性，调整以下参数能辅助提升性能：

• 增大单个Reducer的内存：如果集群资源允许，给处理大分区的Reducer更多内存，避免GC频繁或OOM：

  set mapreduce.reduce.memory.mb=8192; -- 8G内存，根据集群情况调整
  set mapreduce.reduce.java.opts=-Xmx6144m; -- JVM堆内存设为内存的75%左右
  

• 开启PTF优化：Hive的窗口函数依赖PTF（表函数）执行，开启优化能提升处理效率：

  set hive.optimize.ptf=true;
  

• 调整Reducer数量阈值：如果默认的Reducer数量太少，强制增加Reducer来分散负载：

  set hive.exec.reducers.bytes.per.reducer=67108864; -- 每个Reducer处理64M数据，默认256M，改小会增加Reducer数量
  

四、排查并处理异常数据倾斜 #

先找出那些超大分区，看看是否是业务异常导致的：

select upp_sys_id, ssn_seq_all, count(1) as record_cnt
from test.VT_seq_all
group by upp_sys_id, ssn_seq_all
having count(1) > 10000
order by record_cnt desc;

如果发现某些分区是异常数据（比如ssn_seq_all为null、或者测试数据），可以直接过滤掉，或者单独做特殊处理，从根源上消除数据倾斜。

备注：内容来源于stack exchange，提问作者yy zhao