分拆TableSplit让多个mapper同时读取

分拆TableSplit 让多个mapper同时读取

默认情况下，一个region是一个tableSplit，对应一个mapper进行读取，但单mapper读取速度较慢，因此想着把默认一个table split分拆成多个split，这样hadoop就能通过多个mapper读取。 

由于HBase不能像hadoop一样通过以下参数调整split大小，而实现多个mapper读取Java代码  

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

所以目前想到的方法有两种，一是修改TableInputFormatBase，把默认的一个TableSplit分拆成多个，另外一种方法是，通过Coprocessor处理。这里选择修改TableInputFormatBase类。 

HBase权威指南里面有介绍怎么把HBase与MR结合，通过需要用到一下的辅助类实现把HBase表作为数据来源，读取数据： 
Java代码  

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

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

1. public static void initTableMapperJob(byte[] table, Scan scan,  
2.      Classextends TableMapper> mapper,  
3.      Classextends WritableComparable> outputKeyClass,  
4.      Classextends 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代码  

1. public class TableInputFormat extends TableInputFormatBase  
2. implements Configurable   

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

Java代码  

1. public List getSplits(JobContext context) throws IOException {}  

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

1. public List 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 splits = new ArrayList(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代码  

1. public List 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 splits = new ArrayList(  
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 rows = new ArrayList();  
44.             //Iterator  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数。