分拆TableSplit 让多个mapper同时读取

落日映苍穹つ 2022-09-18 13:47 195阅读 0赞

### [分拆TableSplit 让多个mapper同时读取][TableSplit _mapper] ###

默认情况下，一个region是一个tableSplit，对应一个mapper进行读取，但单mapper读取速度较慢，因此想着把默认一个table split分拆成多个split，这样hadoop就能通过多个mapper读取。   
  
由于HBase不能像hadoop一样通过以下参数调整split大小，而实现多个mapper读取

Java代码   ![收藏代码][icon_star.png]

1.  mapred.min.split.size  
2.  mapred.max.split.size

所以目前想到的方法有两种，一是修改TableInputFormatBase，把默认的一个TableSplit分拆成多个，另外一种方法是，通过Coprocessor处理。这里选择修改TableInputFormatBase类。   
  
HBase权威指南里面有介绍怎么把HBase与MR结合，通过需要用到一下的辅助类实现把HBase表作为数据来源，读取数据：

Java代码   ![收藏代码][icon_star.png]

1.  TableMapReduceUtil.initTableMapperJob(table\[0\].getBytes(), scan,  
2.                      UserViewHisMapper2.**class**, Text.**class**, Text.**class**,  
3.                      genRecommendations);

而这个方法，最终是调用以下方法进行初始化设置的：

Java代码   ![收藏代码][icon_star.png]

1.  **public** **static** **void** initTableMapperJob(**byte**\[\] table, Scan scan,  
2.       Class<? **extends** TableMapper> mapper,  
3.       Class<? **extends** WritableComparable> outputKeyClass,  
4.       Class<? **extends** Writable> outputValueClass, Job job,  
5.       **boolean** addDependencyJars)  
6.   **throws** IOException \{  
7.       initTableMapperJob(Bytes.toString(table), scan, mapper, outputKeyClass,  
8.               outputValueClass, job, addDependencyJars, TableInputFormat.**class**);  
9.   \}

所以，思路就应该修改TableInputFormat这个类。而这个类的核心方法是继承了TableInputFormatBase：

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** TableInputFormat **extends** TableInputFormatBase  
2.  **implements** Configurable

最终要修改的则是TableInputFormatBase这个类，修改其以下方法：

Java代码   ![收藏代码][icon_star.png]

1.  **public** List<InputSplit> getSplits(JobContext context) **throws** IOException \{\}

这个方法的核心是，获得table对应所有region的起始row，把每个region作为一个tableSplit：

Java代码   ![收藏代码][icon_star.png]

1.  **public** List<InputSplit> getSplits(JobContext context) **throws** IOException \{  
2.  f (table == **null**) \{  
3.     **throw** **new** IOException("No table was provided.");  
4.    
5.    Pair<**byte**\[\]\[\], **byte**\[\]\[\]> keys = table.getStartEndKeys();  
6.    **if** (keys == **null** || keys.getFirst() == **null** ||  
7.        keys.getFirst().length == 0) \{  
8.      **throw** **new** IOException("Expecting at least one region.");  
9.    \}  
10.   **int** count = 0;  
11.   List<InputSplit> splits = **new** ArrayList<InputSplit>(keys.getFirst().length);  
12.   **for** (**int** i = 0; i < keys.getFirst().length; i++) \{  
13.     **if** ( !includeRegionInSplit(keys.getFirst()\[i\], keys.getSecond()\[i\])) \{  
14.       **continue**;  
15.     \}  
16.     String regionLocation = table.getRegionLocation(keys.getFirst()\[i\]).  
17.       getHostname();  
18.     **byte**\[\] startRow = scan.getStartRow();  
19.     **byte**\[\] stopRow = scan.getStopRow();  
20.     // determine if the given start an stop key fall into the region  
21.     **if** ((startRow.length == 0 || keys.getSecond()\[i\].length == 0 ||  
22.          Bytes.compareTo(startRow, keys.getSecond()\[i\]) < 0) &&  
23.         (stopRow.length == 0 ||  
24.          Bytes.compareTo(stopRow, keys.getFirst()\[i\]) > 0)) \{  
25.       **byte**\[\] splitStart = startRow.length == 0 ||  
26.         Bytes.compareTo(keys.getFirst()\[i\], startRow) >= 0 ?  
27.           keys.getFirst()\[i\] : startRow;  
28.       **byte**\[\] splitStop = (stopRow.length == 0 ||  
29.         Bytes.compareTo(keys.getSecond()\[i\], stopRow) <= 0) &&  
30.         keys.getSecond()\[i\].length > 0 ?  
31.           keys.getSecond()\[i\] : stopRow;  
32.       InputSplit split = **new** TableSplit(table.getTableName(),  
33.         splitStart, splitStop, regionLocation);  
34.       splits.add(split);  
35.       **if** (LOG.isDebugEnabled())  
36.         LOG.debug("getSplits: split -> " + (count++) + " -> " + split);  
37.     \}  
38.   \}  
39.   **return** splits;  
40. \}

这里要做的就是，把本来属于一个tableSplit的row在细分，分成自己希望的多个小split。但没有找到轻巧的实现，唯有不断迭代，把一个tableSplit的row全部取出，再拆分了，有点蛮力。   
以下是我的实现方法：

Java代码   ![收藏代码][icon_star.png]

1.  **public** List<InputSplit> getSplits(JobContext context) **throws** IOException \{  
2.      **if** (table == **null**) \{  
3.          **throw** **new** IOException("No table was provided.");  
4.      \}  
5.      Pair<**byte**\[\]\[\], **byte**\[\]\[\]> keys = table.getStartEndKeys();  
6.      **if** (keys == **null** || keys.getFirst() == **null**  
7.              || keys.getFirst().length == 0) \{  
8.          **throw** **new** IOException("Expecting at least one region.");  
9.      \}  
10.     **int** count = 0;  
11.     List<InputSplit> splits = **new** ArrayList<InputSplit>(  
12.             keys.getFirst().length);  
13.     **for** (**int** i = 0; i < keys.getFirst().length; i++) \{  
14.         **if** (!includeRegionInSplit(keys.getFirst()\[i\], keys.getSecond()\[i\])) \{  
15.             **continue**;  
16.         \}  
17.         String regionLocation = table.getRegionLocation(keys.getFirst()\[i\],**true**)  
18.                 .getHostname();  
19.         **byte**\[\] startRow = scan.getStartRow();  
20.         **byte**\[\] stopRow = scan.getStopRow();  
21.         // determine if the given start an stop key fall into the region  
22.         **if** ((startRow.length == 0 || keys.getSecond()\[i\].length == 0 || Bytes  
23.                 .compareTo(startRow, keys.getSecond()\[i\]) < 0)  
24.                 && (stopRow.length == 0 || Bytes.compareTo(stopRow,  
25.                         keys.getFirst()\[i\]) > 0)) \{  
26.             **byte**\[\] splitStart = startRow.length == 0  
27.                     || Bytes.compareTo(keys.getFirst()\[i\], startRow) >= 0 ? keys  
28.                     .getFirst()\[i\] : startRow;  
29.             **byte**\[\] splitStop = (stopRow.length == 0 || Bytes.compareTo(  
30.                     keys.getSecond()\[i\], stopRow) <= 0)  
31.                     && keys.getSecond()\[i\].length > 0 ? keys.getSecond()\[i\]  
32.                     : stopRow;  
33.   
34.             Scan scan1 = **new** Scan();  
35.             scan1.setStartRow(splitStart);  
36.             scan1.setStopRow(splitStop);  
37.             scan1.setFilter(**new** KeyOnlyFilter());  
38.             scan1.setBatch(500);  
39.               
40.             ResultScanner resultscanner = table.getScanner(scan1);  
41.               
42.             //用来保存该region的所有key  
43.             List<String> rows = **new** ArrayList<String>();  
44.             //Iterator<Result>  it = resultscanner.iterator();  
45.               
46.             **for**(Result rs : resultscanner)  
47.             \{  
48.                 **if**(rs.isEmpty())  
49.                     **continue**;  
50.                 rows.add(**new** String(rs.getRow()));  
51.             \}  
52.               
53.             **int** splitSize = rows.size() / mappersPerSplit;  
54.               
55.             **for** (**int** j = 0; j < mappersPerSplit; j++) \{  
56.                 TableSplit tablesplit = **null**;  
57.                 **if** (j == mappersPerSplit - 1)  
58.                     tablesplit = **new** TableSplit(table.getTableName(),  
59.                             rows.get(j \* splitSize).getBytes(),  
60.                             rows.get(rows.size() - 1).getBytes(),  
61.                             regionLocation);  
62.                 **else**  
63.                     tablesplit = **new** TableSplit(table.getTableName(),  
64.                             rows.get(j \* splitSize).getBytes(),  
65.                             rows.get(j \* splitSize + splitSize).getBytes(), regionLocation);  
66.                 splits.add(tablesplit);  
67.                 **if** (LOG.isDebugEnabled())  
68.                     LOG.debug((**new** StringBuilder())  
69.                             .append("getSplits: split -> ").append(i++)  
70.                             .append(" -> ").append(tablesplit).toString());  
71.             \}  
72.             resultscanner.close();                
73.         \}  
74.     \}  
75.     **return** splits;  
76. \}

通过配置设置需要拆分的split数。

[TableSplit _mapper]: http://blackwing.iteye.com/blog/1763961
[icon_star.png]: /images/20220828/17761ac2e62841d7809bdbcdb8b3b207.png