hive map-reduce個數及合並小文件

1. map數計算方式

long splitSize = computeSplitSize(blockSize, minSize, maxSize);
protected long computeSplitSize(long blockSize, long minSize,  
                                long maxSize) {  
  return Math.max(minSize, Math.min(maxSize, blockSize));  
}  

/******************************************************/
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));

protected long getFormatMinSplitSize() {  
    return 1;  
  }  

public static long getMinSplitSize(JobContext job) {  
    return job.getConfiguration().getLong(SPLIT_MINSIZE, 1L);  
  }  

public static final String SPLIT_MINSIZE =   
    "mapreduce.input.fileinputformat.split.minsize"; //默認值為0，所以默認minSize返回1

/******************************************************/
long maxSize = getMaxSplitSize(job);

public static long getMaxSplitSize(JobContext context) {  
    return context.getConfiguration().getLong(SPLIT_MAXSIZE,   
                                              Long.MAX_VALUE);  
  }  

 public static final String SPLIT_MAXSIZE =   
    "mapreduce.input.fileinputformat.split.maxsize";//默認maxSize返回Long.MAX_VALUE

/******************************************************/   
blockSize為配置文件中的參數dfs.block.size，默認值為128M

 

2. 影響map個數的因素

input的文件總個數,
input的文件大小,
集群設置的文件塊大小(目前為128M, 可在Hive中通過set dfs.block.size;命令查看到，該參數不能自定義修改)；

舉例： 
a)假設input目錄下有1個文件a,大小為780M,那么Hadoop會將該文件a分隔成7個塊（6個128m的塊和1個12m的塊），從而產生7個map數
b)假設input目錄下有3個文件a,b,c,大小分別為10m，20m，130m，那么hadoop會分隔成4個塊（10m,20m,128m,2m）,從而產生4個map數

即，如果文件大於塊大小(128m),那么會拆分，如果小於塊大小，則把該文件當成一個塊

 

3.修改map個數

3.1 合並小文件減少map數

a)輸入合並。即在Map前合並小文件
set mapred.min.split.size=100000000;
set mapred.max.split.size=100000000;
-- 一個節點上split的至少的大小 ，決定了多個data node上的文件是否需要合並，不知道怎么用
set mapred.min.split.size.per.node=50000000;
-- 一個交換機下split的至少的大小，決定了多個交換機上的文件是否需要合並，不知道怎么用
set mapred.min.split.size.per.rack=50000000;
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

100000000表示100M
set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;這個參數表示執行前進行小文件合並

過期的對應配置可以查詢：https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/DeprecatedProperties.html

mapred.min.split.size相當於mapreduce.input.fileinputformat.split.minsize，設置哪一個都一樣
mapred.max.split.size相當於mapreduce.input.fileinputformat.split.maxsize，設置哪一個都一樣
所以根據上面求splitSize的公式，可以知道，需增加mapper數，則減少這些值；需減少mapper數，則增加這些值，一般就配置這兩個值就行了。

b)Hive結果合並
hive.merge.mapfiles 在map-only job后合並文件，默認true
hive.merge.mapredfiles 在map-reduce job后合並文件，默認false
hive.merge.size.per.task 合並后每個文件的大小，默認256000000
hive.merge.smallfiles.avgsize 平均文件大小，是決定是否執行合並操作的閾值，默認16000000

Hive在對結果文件進行合並時會執行一個額外的map-only腳本，mapper的數量是文件總大小除以size.per.task參數所得的值，觸發合並的條件是：

 

3.2 增加map個數

當一個文件大小不大（假設120m，塊大小為128m），但是好多數據（好多行），那么起了一個map處理將特別慢，此時需要增加map數來加快處理
此時可設置:
set mapred.map.tasks=10;
map的最終個數取決於
goalsize=min(totalsize/mapred.reduce.tasks,塊大小)
goalsize就相當於splitsize

假設：
input有3個300M的文件，塊大小為256M

場景1：
不設置map數，那么會有6個map，分別處理256M 44M 256M 44M 256M 44M
場景2：
map數為2個，goalsize=((900M)/2, 塊大小)=256M，map數和處理的數據和場景1一樣
場景3：
map數為6個，goalsize=((900M)/6, 塊大小)=150，分別處理150 150 150 150 150 150

場景3分布比較均勻合理，不用等待大文件處理完成

 

4.默認reduce計算方式

hive.exec.reducers.bytes.per.reducer（每個reduce任務處理的數據量，默認為1000^3=1G） 
hive.exec.reducers.max（每個任務最大的reduce數，默認為999）

計算reducer數的公式：N=min(參數2，總輸入數據量/參數1) 即，如果reduce的輸入（map的輸出，一個map對應一個reduce?）總大小不超過1G,那么只會有一個reduce任務；

 

5.調整reduce個數

#在設置reduce個數的時候也需要考慮這兩個原則：使大數據量利用合適的reduce數；使單個reduce任務處理合適的數據量；

#調整hive.exec.reducers.bytes.per.reducer參數的值；
set hive.exec.reducers.bytes.per.reducer=500000000; （500M）

#調整hive.exec.reducers.max
set hive.exec.reducers.max=15

#設置reduce大小
set mapred.reduce.tasks = 15;

 

6.只有一個reduce的情況

• 沒有group by的匯總，比如把select pt,count(1) from popt_tbaccountcopy_mes where pt = '2012-07-04' group by pt; 寫成 select count(1) from popt_tbaccountcopy_mes where pt = '2012-07-04';  這點非常常見，希望大家盡量改寫。
• 用了Order by
• 有笛卡爾積

通常這些情況下，除了找辦法來變通和避免，我暫時沒有什么好的辦法，因為這些操作都是全局的，所以hadoop不得不用一個reduce去完成；

7.設置map reduce輸出參數合並小文件

--設置map端輸出進行合並，默認為true  
set hive.merge.mapfiles = true  
--設置reduce端輸出進行合並，默認為false  
set hive.merge.mapredfiles = true  
--設置合並文件的大小  
set hive.merge.size.per.task = 256*1000*1000  
--當輸出文件的平均大小小於該值時，啟動一個獨立的MapReduce任務進行文件merge。  
set hive.merge.smallfiles.avgsize=16000000 

 

8.參考鏈接

1. http://blog.csdn.net/longzilong216/article/details/20711433
2. http://www.makaidong.com/%E5%8D%9A%E5%AE%A2%E5%9B%AD%E6%8E%92%E8%A1%8C/16451.shtml
3. http://younglibin.iteye.com/blog/1929255
4. http://blog.csdn.net/yycdaizi/article/details/43341239
5. https://github.com/linkedin/dr-elephant/wiki/Tuning-Tips
6. http://superlxw1234.iteye.com/blog/1582880
7. http://www.cnblogs.com/thinkpad/p/5173641.html
8. https://blog.csdn.net/kanbuqinghuanyizhang/article/details/79132564