码迷,mamicode.com
首页 > 其他好文 > 详细

TableInputFormat分片及分片数据读取源码级分析

时间:2014-07-07 20:09:05      阅读:260      评论:0      收藏:0      [点我收藏+]

标签:style   blog   http   color   使用   数据   

  我们在MapReduce中TextInputFormat分片和读取分片数据源码级分析 这篇中以TextInputFormat为例讲解了InputFormat的分片过程以及RecordReader读取分片数据的过程。接下来咱们分析TableInputFormat的分片信息和数据读取过程。

  TableInputFormat这是专门处理基于HBase的MapReduce的输入数据的格式类。我们可以看看继承结构:(1)public class TableInputFormat extends TableInputFormatBase implements Configurable;(2)public abstract class TableInputFormatBase extends InputFormat<ImmutableBytesWritable, Result>。其中InputFormat是输入格式的基类。

  TableInputFormat类主要是构造HTable对象和Scan对象,主要在方法setConf(Configuration configuration)构造,代码如下:

bubuko.com,布布扣
 1 /**
 2    * Sets the configuration. This is used to set the details for the table to
 3    * be scanned.
 4    *
 5    * @param configuration  The configuration to set.
 6    * @see org.apache.hadoop.conf.Configurable#setConf(
 7    *   org.apache.hadoop.conf.Configuration)
 8    */
 9   @Override
10   public void setConf(Configuration configuration) {
11     this.conf = configuration;
12     String tableName = conf.get(INPUT_TABLE);
13     try {
14       setHTable(new HTable(new Configuration(conf), tableName));
15     } catch (Exception e) {
16       LOG.error(StringUtils.stringifyException(e));
17     }
18 
19     Scan scan = null;
20 
21     if (conf.get(SCAN) != null) {
22       try {
23         scan = TableMapReduceUtil.convertStringToScan(conf.get(SCAN));
24       } catch (IOException e) {
25         LOG.error("An error occurred.", e);
26       }
27     } else {
28       try {
29         scan = new Scan();
30 
31         if (conf.get(SCAN_ROW_START) != null) {
32           scan.setStartRow(Bytes.toBytes(conf.get(SCAN_ROW_START)));
33         }
34 
35         if (conf.get(SCAN_ROW_STOP) != null) {
36           scan.setStopRow(Bytes.toBytes(conf.get(SCAN_ROW_STOP)));
37         }
38 
39         if (conf.get(SCAN_COLUMNS) != null) {
40           addColumns(scan, conf.get(SCAN_COLUMNS));
41         }
42 
43         if (conf.get(SCAN_COLUMN_FAMILY) != null) {
44           scan.addFamily(Bytes.toBytes(conf.get(SCAN_COLUMN_FAMILY)));
45         }
46 
47         if (conf.get(SCAN_TIMESTAMP) != null) {
48           scan.setTimeStamp(Long.parseLong(conf.get(SCAN_TIMESTAMP)));
49         }
50 
51         if (conf.get(SCAN_TIMERANGE_START) != null && conf.get(SCAN_TIMERANGE_END) != null) {
52           scan.setTimeRange(
53               Long.parseLong(conf.get(SCAN_TIMERANGE_START)),
54               Long.parseLong(conf.get(SCAN_TIMERANGE_END)));
55         }
56 
57         if (conf.get(SCAN_MAXVERSIONS) != null) {
58           scan.setMaxVersions(Integer.parseInt(conf.get(SCAN_MAXVERSIONS)));
59         }
60 
61         if (conf.get(SCAN_CACHEDROWS) != null) {
62           scan.setCaching(Integer.parseInt(conf.get(SCAN_CACHEDROWS)));
63         }
64 
65         // false by default, full table scans generate too much BC churn
66         scan.setCacheBlocks((conf.getBoolean(SCAN_CACHEBLOCKS, false)));
67       } catch (Exception e) {
68           LOG.error(StringUtils.stringifyException(e));
69       }
70     }
71 
72     setScan(scan);
73   }
View Code

  首先会通过配置信息获取HBase表名,然后构造一个HTable对象;如果用户自己的作业中有配置Scan的话,就会解码Scan字符串转换为Scan对象;如果用户没配置Scan就会创建一个默认的Scan,进行一些基本配置。

  关于TableInputFormatBase,我们重点还是讲两个方法:RecordReader<ImmutableBytesWritable, Result> createRecordReader(InputSplit split, TaskAttemptContext context)方法和List<InputSplit> getSplits(JobContext context)方法,前者是读取分片信息所指的数据供TableMapper处理,后者是构造HBase表的分片信息。这里的分片信息是TableSplit extends InputSplit implements Writable, Comparable,这个TableSpli维护4个字段,HBase表名:byte [] tableName、scan起始rowkey:byte [] startRow、scan结束rowkey:byte [] endRow、以及该region所在节点:String regionLocation。

  一、先看getSplits方法吧,这里一个split一般对应一个完整region,除非用户设定的开始和结束rowkey不是region的边界,代码如下:

bubuko.com,布布扣
 1   /**
 2    * Calculates the splits that will serve as input for the map tasks. The
 3    * number of splits matches the number of regions in a table.
 4    *
 5    * @param context  The current job context.
 6    * @return The list of input splits.
 7    * @throws IOException When creating the list of splits fails.
 8    * @see org.apache.hadoop.mapreduce.InputFormat#getSplits(
 9    *   org.apache.hadoop.mapreduce.JobContext)
10    */
11   @Override
12   public List<InputSplit> getSplits(JobContext context) throws IOException {
13     if (table == null) {
14         throw new IOException("No table was provided.");
15     }
16     // Get the name server address and the default value is null.
17     this.nameServer =
18       context.getConfiguration().get("hbase.nameserver.address", null);
19     
20     Pair<byte[][], byte[][]> keys = table.getStartEndKeys();//获取所有Region的开始rowkey和结束rowkey
21     if (keys == null || keys.getFirst() == null ||
22         keys.getFirst().length == 0) {    //
23         //table的第一个region的startKey必须是EMPTY_BYTE_ARRAY,否则输出FIRST_REGION_STARTKEY_NOT_EMPTY信息 
24       HRegionLocation regLoc = table.getRegionLocation(
25           HConstants.EMPTY_BYTE_ARRAY, false);
26       if (null == regLoc) {    //一个region也没有
27         throw new IOException("Expecting at least one region.");
28       }
29       List<InputSplit> splits = new ArrayList<InputSplit>(1);
30       //构造一个TableSplit,起始rowkey和结束rowkey都是EMPTY_BYTE_ARRAY
31       InputSplit split = new TableSplit(table.getTableName(),
32           HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, regLoc
33               .getHostnamePort().split(Addressing.HOSTNAME_PORT_SEPARATOR)[0]);
34       splits.add(split);
35       return splits;        //返回分片信息
36     }
37     List<InputSplit> splits = new ArrayList<InputSplit>(keys.getFirst().length);
38     for (int i = 0; i < keys.getFirst().length; i++) {    //有多个region
39       if ( !includeRegionInSplit(keys.getFirst()[i], keys.getSecond()[i])) {    //这个方法一直返回true
40         continue;
41       }
42       HServerAddress regionServerAddress = 
43         table.getRegionLocation(keys.getFirst()[i]).getServerAddress();
44       InetAddress regionAddress =
45         regionServerAddress.getInetSocketAddress().getAddress();//获取region所在地址
46       String regionLocation;
47       try {
48         regionLocation = reverseDNS(regionAddress);//将地址转换为字符串的主机名
49       } catch (NamingException e) {
50         LOG.error("Cannot resolve the host name for " + regionAddress +
51             " because of " + e);
52         regionLocation = regionServerAddress.getHostname();
53       }
54 
55             byte[] startRow = scan.getStartRow();    //获取scan的开始rowkey
56             byte[] stopRow = scan.getStopRow();    //获取scan的结束rowkey
57             // determine if the given start an stop key fall into the region
58             //比较用户设定的rowkey范围在那些region之中
59       if ((startRow.length == 0 || keys.getSecond()[i].length == 0 ||
60                      Bytes.compareTo(startRow, keys.getSecond()[i]) < 0) &&
61           (stopRow.length == 0 ||
62            Bytes.compareTo(stopRow, keys.getFirst()[i]) > 0)) {
63         byte[] splitStart = startRow.length == 0 ||
64           Bytes.compareTo(keys.getFirst()[i], startRow) >= 0 ?
65             keys.getFirst()[i] : startRow;
66         byte[] splitStop = (stopRow.length == 0 ||
67           Bytes.compareTo(keys.getSecond()[i], stopRow) <= 0) &&
68           keys.getSecond()[i].length > 0 ?
69             keys.getSecond()[i] : stopRow;
70         InputSplit split = new TableSplit(table.getTableName(),
71           splitStart, splitStop, regionLocation);    //构造TableSplit
72         splits.add(split);
73         if (LOG.isDebugEnabled()) {
74           LOG.debug("getSplits: split -> " + i + " -> " + split);
75         }
76       }
77     }
78     return splits;
79   }
View Code

   Pair这个结构由两部分组成:start和end,都是byte[][],这是一个二维byte数组,其中一维是region的顺序,二维是start或者end的字符序列。通过getFirst方法就可以获取所有region的start rowkey,通过getSecond可以获得所有的end rowkey。

  这里还有一个要注意的就是每个HBase表的第一个region是没有start rowkey,最后一个region是没有end  rowkey,这里的“没有”是指的是table.getStartEndKeys()这个方法获取的结果,另外大伙也可以从WEB UI中查看指定的HBase表的region信息也可以看到第一个region的start和最后一个region的end并没有显示。还有就是如果这个HBase表只有一个region的话,getFirst方法返回是没有数据的;getSecond也没数据。

  (1)、getSplits方法中的第一个if语句段是上面说的只有一个region的情况。每个HBase表的第一个region的start rowkey都是EMPTY_BYTE_ARRAY,这是一个长度为0的byte数组。table.getRegionLocation方法会找指定的rowkey所在的region所在信息HRegionLocation。如果只有一个region的话,就先找EMPTY_BYTE_ARRAY所在region信息,如果没有这样的信息就是出现了错误;如果如果有这样的信息的话,就构建一个长度为1的InputSplit列表splits,构造一个TableSplit:设定HBase表名,scan起始和结束rowkey都是EMPTY_BYTE_ARRAY,再加上region所在节点,把这个TableSplit加入splits,返回这个splits。

  (2)、接下来就是HBase表有多个region的情况,构建长度为keys.getFirst().length的InputSplit列表,然后遍历keys.getFirst()获取每个region的位置信息并将其转换成String类型(reverseDNS方法从reverseDNSCacheMap中获取,reverseDNSCacheMap是存储IPAddress => HostName的映射);然后获取用户设定的起始和结束rowkey,并和当前的region的起始和比较结束,如果有rowkey包含这个region就会将这region当做一个InputSplit放入列表中,最后待遍历完之后返回split列表。判断当前region是否应该加入InputSplit列表的条件就是循环中的最后一个if语句段,条件是:((startRow.length == 0 || keys.getSecond()[i].length == 0 ||Bytes.compareTo (startRow, keys.getSecond()[i]) < 0) && (stopRow.length == 0 || Bytes.compareTo(stopRow, keys.getFirst()[i]) > 0)),分解这个条件为两部分:A、(startRow.length == 0 || keys.getSecond()[i].length == 0 ||Bytes.compareTo (startRow, keys.getSecond()[i]) < 0)这个是要确定用户设定的start rowkey是否小于当前region的结束rowkey;B、(stopRow.length == 0 || Bytes.compareTo(stopRow, keys.getFirst()[i]) > 0)这个是要确定用户设定的end rowkey是否大于当前region的开始rowkey,这俩条件必须同时满足才可以,一旦满足就要确定这个TableSplit的开始rowkey和结束rowkey了:A、startRow.length == 0 || Bytes.compareTo(keys.getFirst()[i], startRow) >= 0如果为true的话,说明用户设定的起始rowkey可能是从表的开头开始或者是当前region的起始rowkey大于用户设定的则应该将当前region的起始rowkey作为TableSplit的起始rowkey,如果表达式为false的话将用户设定的起始rowkey作为TableSplit的起始rowkey;B、(stopRow.length == 0 || Bytes.compareTo(keys.getSecond()[i], stopRow) <= 0) && keys.getSecond()[i].length > 0如果为true的话,说明首先要确定是否设定的结束rowkey或者当前region的结束rowkey小于用户设定的结束rowkey,且要保证当前region不是最后一个(keys.getSecond()[i].length > 0),这样的的TableSplit的结束rowkey就是当前region的结束rowkey,如果为false则将用户设定的结束rowkey为TableSplit的结束rowkey,为什么要不是最后一个region呢?因为最后一个region的end rowkey的长度始终为0,比较之下会将最后一个region的end rowkey设置给TableSplit,显然这是不对,能到这里说明这个region应该被分配给一个TableSplit,如果是最后一个region的话,那么这个TableSplit的结束rowkey应该是用户设定的而非这个region自己的。获得这个tableSplit的开始和结束rowkey之后就可以封装这个TableSplit了,并放入InputSplit列表。最终待所有的region遍历结束之后返回这个InputSplit列表。

  这样getSplits方法就结束了。

  二、createRecordReader方法,这个方法代码如下:

bubuko.com,布布扣
 1 /**
 2    * Builds a TableRecordReader. If no TableRecordReader was provided, uses
 3    * the default.
 4    *
 5    * @param split  The split to work with.
 6    * @param context  The current context.
 7    * @return The newly created record reader.
 8    * @throws IOException When creating the reader fails.
 9    * @see org.apache.hadoop.mapreduce.InputFormat#createRecordReader(
10    *   org.apache.hadoop.mapreduce.InputSplit,
11    *   org.apache.hadoop.mapreduce.TaskAttemptContext)
12    */
13   @Override
14   public RecordReader<ImmutableBytesWritable, Result> createRecordReader(
15       InputSplit split, TaskAttemptContext context)
16   throws IOException {
17     if (table == null) {
18       throw new IOException("Cannot create a record reader because of a" +
19           " previous error. Please look at the previous logs lines from" +
20           " the task‘s full log for more details.");
21     }
22     TableSplit tSplit = (TableSplit) split;
23     TableRecordReader trr = this.tableRecordReader;
24     // if no table record reader was provided use default
25     if (trr == null) {
26       trr = new TableRecordReader();
27     }
28     Scan sc = new Scan(this.scan);
29     sc.setStartRow(tSplit.getStartRow());
30     sc.setStopRow(tSplit.getEndRow());
31     trr.setScan(sc);
32     trr.setHTable(table);
33     try {
34       trr.initialize(tSplit, context);
35     } catch (InterruptedException e) {
36       throw new InterruptedIOException(e.getMessage());
37     }
38     return trr;
39   }
View Code

  这个方法主要是获取TableSplit,然后构造一个Scan,设定开始和结束rowkey;设定HTablePool;将Scan和HTable传递给一个TableRecordReader对象,然后调用initialize(tSplit, context)初始化,最后返回这个TableRecordReader。可以看出TableRecordReader这个是读取key/value的。TableRecordReader中实际操作数据的是TableRecordReaderImpl,TableRecordReader的nextKeyValue()、getCurrentValue()、initialize、getCurrentKey()方法会调用TableRecordReaderImpl的相应方法。

  TableRecordReaderImpl的initialize方法主要是重新创建一个新的Scan,并将createRecordReader传过来的赋值给这个新的currentScan,并获取对应的ResultScanner。

  TableRecordReaderImpl的nextKeyValue()会先创建一个key = new ImmutableBytesWritable()和value = new Result(),这就是我们继承TableMapper中map方法中的参数类型,然后每次调用该方法通过执行value = this.scanner.next()方法来获取HBase中的一行数据赋值给value,这里表明如果scanner.next()运行无异常的话key中是没有数据的(出现异常之后会存储value对应行的rowkey),只有value有数据。执行了这个方法就可以通过getCurrentValue()、getCurrentKey()方法来获取value和key了。

 

  上面这些讲解了在HBase上使用MapReduce时的分片过程及如何读取这些分片上的数据的。

TableInputFormat分片及分片数据读取源码级分析,布布扣,bubuko.com

TableInputFormat分片及分片数据读取源码级分析

标签:style   blog   http   color   使用   数据   

原文地址:http://www.cnblogs.com/lxf20061900/p/3812812.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!