Hive INSERT OVERWRITE单分区生成大文件，如何拆分至500MB？

背景与问题 #

执行Hive INSERT OVERWRITE查询，将两张表关联后的结果写入按CPD_DT和CUT_NO分区的AVRO表CS.STAGING_AVRO，但每个分区仅生成1个3GB的大文件，需拆分为多个500MB左右的文件。

原查询与配置 #

SET hive.merge.mapredfiles=false;
SET mapred.reduce.tasks=${hiveconf:NUM_REDUCERS};
SET hive.mapred.supports.subdirectories=TRUE;
SET mapred.input.dir.recursive=TRUE;
SET hive.execution.engine=tez;
SET hive.tez.container.size=6144;

INSERT OVERWRITE TABLE CS.STAGING_AVRO PARTITION (CPD_DT='${hiveconf:CPD_DT}',CUT_NO='${hiveconf:CUT_NO}')
SELECT DISTINCT CS_TRAN.field1, CS_TRAN.field2 .... 
FROM ( SELECT A, B, C, ... FROM NMS1.TBL1 WHERE ... ) CS_TRAN
JOIN NMS2.TBL2 ON ....

问题现象 #

每个分区生成单个3GB文件：

• /data/cpd_dt=20230220/cut_no=00/000000_0 → 3GB
• /data/cpd_dt=20230220/cut_no=01/000000_0 → 3GB
• /data/cpd_dt=20230220/cut_no=02/000000_0 → 3GB

已尝试但无效的配置 #

SET mapred.max.split.size=1000000;
SET hive.exec.reducers.bytes.per.reducer=1000000;
SET hive.merge.size.per.task=1000000;
SET hive.merge.smallfiles.avgsize=1000000;
SET hive.exec.compress.output=true;

可行解决方法 #

当前使用Tez引擎，之前的配置多针对MapReduce，且参数值设置不合理（如1000000为1MB，远小于目标500MB），以下是针对性调整：

1. 调整Tez的Reducer数量控制参数 #

替换MapReduce旧参数，用Tez兼容的参数控制Reducer数量：

-- 设置每个Reducer处理的字节数（500MB = 536870912字节）
SET hive.exec.reducers.bytes.per.reducer=536870912;
-- 可选：自动计算不生效时，强制指定Reducer数量（3GB/500MB≈6个）
SET mapreduce.job.reduces=6;

2. 开启Tez的输出文件合并 #

原配置关闭了MapReduce的文件合并，需开启Tez专属的合并参数：

SET hive.merge.tezfiles=true; -- 开启Tez输出文件合并
SET hive.merge.size.per.task=536870912; -- 每个合并任务处理的目标文件大小
SET hive.merge.smallfiles.avgsize=268435456; -- 当平均文件大小小于256MB时触发合并

3. 解决DISTINCT导致的单Reducer瓶颈 #

DISTINCT易导致数据在单个Reducer聚合，需改为分布式去重：

• 先局部去重再全局去重：

  INSERT OVERWRITE TABLE CS.STAGING_AVRO PARTITION (CPD_DT='${hiveconf:CPD_DT}',CUT_NO='${hiveconf:CUT_NO}')
  SELECT DISTINCT field1, field2, ...
  FROM (
      SELECT DISTINCT CS_TRAN.field1, CS_TRAN.field2, ...
      FROM (SELECT A, B, C, ... FROM NMS1.TBL1 WHERE ... ) CS_TRAN
      JOIN NMS2.TBL2 ON ....
  ) tmp;
  

• 或用DISTRIBUTE BY强制散列数据到多个Reducer：

  INSERT OVERWRITE TABLE CS.STAGING_AVRO PARTITION (CPD_DT='${hiveconf:CPD_DT}',CUT_NO='${hiveconf:CUT_NO}')
  SELECT DISTINCT field1, field2, ...
  FROM (
      SELECT CS_TRAN.field1, CS_TRAN.field2, ...
      FROM (SELECT A, B, C, ... FROM NMS1.TBL1 WHERE ... ) CS_TRAN
      JOIN NMS2.TBL2 ON ....
  ) tmp
  DISTRIBUTE BY field1; -- 选择基数较高的字段，确保数据均匀分布
  

4. 优化AVRO输出配置 #

开启压缩减少文件体积，同时确保无强制单文件设置：

SET hive.exec.compress.output=true;
SET avro.output.codec=snappy; -- 平衡压缩比与读写速度

注意事项 #

• 所有参数需放在INSERT语句之前生效
• 强制指定Reducer数量时，需根据实际数据总量调整
• DISTRIBUTE BY的字段需选择基数足够大的，避免数据倾斜

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