1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mapreduce;
20
21 import java.io.Closeable;
22 import java.io.IOException;
23 import java.net.InetAddress;
24 import java.net.InetSocketAddress;
25 import java.net.UnknownHostException;
26 import java.util.ArrayList;
27 import java.util.Arrays;
28 import java.util.HashMap;
29 import java.util.List;
30
31 import javax.naming.NamingException;
32
33 import org.apache.commons.logging.Log;
34 import org.apache.commons.logging.LogFactory;
35 import org.apache.hadoop.conf.Configuration;
36 import org.apache.hadoop.hbase.classification.InterfaceAudience;
37 import org.apache.hadoop.hbase.classification.InterfaceStability;
38 import org.apache.hadoop.hbase.HConstants;
39 import org.apache.hadoop.hbase.HRegionLocation;
40 import org.apache.hadoop.hbase.TableName;
41 import org.apache.hadoop.hbase.client.Admin;
42 import org.apache.hadoop.hbase.client.Connection;
43 import org.apache.hadoop.hbase.client.ConnectionFactory;
44 import org.apache.hadoop.hbase.client.HTable;
45 import org.apache.hadoop.hbase.client.NeedUnmanagedConnectionException;
46 import org.apache.hadoop.hbase.client.RegionLocator;
47 import org.apache.hadoop.hbase.client.Result;
48 import org.apache.hadoop.hbase.client.Scan;
49 import org.apache.hadoop.hbase.client.Table;
50 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
51 import org.apache.hadoop.hbase.util.Addressing;
52 import org.apache.hadoop.hbase.util.Bytes;
53 import org.apache.hadoop.hbase.util.Pair;
54 import org.apache.hadoop.hbase.util.RegionSizeCalculator;
55 import org.apache.hadoop.hbase.util.Strings;
56 import org.apache.hadoop.mapreduce.InputFormat;
57 import org.apache.hadoop.mapreduce.InputSplit;
58 import org.apache.hadoop.mapreduce.JobContext;
59 import org.apache.hadoop.mapreduce.RecordReader;
60 import org.apache.hadoop.mapreduce.TaskAttemptContext;
61 import org.apache.hadoop.net.DNS;
62 import org.apache.hadoop.util.StringUtils;
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104 @InterfaceAudience.Public
105 @InterfaceStability.Stable
106 public abstract class TableInputFormatBase
107 extends InputFormat<ImmutableBytesWritable, Result> {
108
109
110 public static final String MAPREDUCE_INPUT_AUTOBALANCE = "hbase.mapreduce.input.autobalance";
111
112
113 public static final String INPUT_AUTOBALANCE_MAXSKEWRATIO = "hbase.mapreduce.input.autobalance" +
114 ".maxskewratio";
115
116
117 public static final String TABLE_ROW_TEXTKEY = "hbase.table.row.textkey";
118
119 private static final Log LOG = LogFactory.getLog(TableInputFormatBase.class);
120
121 private static final String NOT_INITIALIZED = "The input format instance has not been properly " +
122 "initialized. Ensure you call initializeTable either in your constructor or initialize " +
123 "method";
124 private static final String INITIALIZATION_ERROR = "Cannot create a record reader because of a" +
125 " previous error. Please look at the previous logs lines from" +
126 " the task's full log for more details.";
127
128
129
130
131 private Scan scan = null;
132
133 private Admin admin;
134
135 private Table table;
136
137 private RegionLocator regionLocator;
138
139 private TableRecordReader tableRecordReader = null;
140
141 private Connection connection;
142
143
144
145 private HashMap<InetAddress, String> reverseDNSCacheMap =
146 new HashMap<InetAddress, String>();
147
148
149
150
151
152
153
154
155
156
157
158
159
160 @Override
161 public RecordReader<ImmutableBytesWritable, Result> createRecordReader(
162 InputSplit split, TaskAttemptContext context)
163 throws IOException {
164
165 if (table == null) {
166 initialize(context);
167 }
168
169 try {
170 if (getTable() == null) {
171
172 throw new IOException(INITIALIZATION_ERROR);
173 }
174 } catch (IllegalStateException exception) {
175 throw new IOException(INITIALIZATION_ERROR, exception);
176 }
177 TableSplit tSplit = (TableSplit) split;
178 LOG.info("Input split length: " + StringUtils.humanReadableInt(tSplit.getLength()) + " bytes.");
179 final TableRecordReader trr =
180 this.tableRecordReader != null ? this.tableRecordReader : new TableRecordReader();
181 Scan sc = new Scan(this.scan);
182 sc.setStartRow(tSplit.getStartRow());
183 sc.setStopRow(tSplit.getEndRow());
184 trr.setScan(sc);
185 trr.setTable(getTable());
186 return new RecordReader<ImmutableBytesWritable, Result>() {
187
188 @Override
189 public void close() throws IOException {
190 trr.close();
191 closeTable();
192 }
193
194 @Override
195 public ImmutableBytesWritable getCurrentKey() throws IOException, InterruptedException {
196 return trr.getCurrentKey();
197 }
198
199 @Override
200 public Result getCurrentValue() throws IOException, InterruptedException {
201 return trr.getCurrentValue();
202 }
203
204 @Override
205 public float getProgress() throws IOException, InterruptedException {
206 return trr.getProgress();
207 }
208
209 @Override
210 public void initialize(InputSplit inputsplit, TaskAttemptContext context) throws IOException,
211 InterruptedException {
212 trr.initialize(inputsplit, context);
213 }
214
215 @Override
216 public boolean nextKeyValue() throws IOException, InterruptedException {
217 return trr.nextKeyValue();
218 }
219 };
220 }
221
222 protected Pair<byte[][],byte[][]> getStartEndKeys() throws IOException {
223 return getRegionLocator().getStartEndKeys();
224 }
225
226
227
228
229
230
231
232
233
234
235
236 @Override
237 public List<InputSplit> getSplits(JobContext context) throws IOException {
238 boolean closeOnFinish = false;
239
240
241 if (table == null) {
242 initialize(context);
243 closeOnFinish = true;
244 }
245
246
247 try {
248 if (getTable() == null) {
249
250 throw new IOException(INITIALIZATION_ERROR);
251 }
252 } catch (IllegalStateException exception) {
253 throw new IOException(INITIALIZATION_ERROR, exception);
254 }
255
256 try {
257 RegionSizeCalculator sizeCalculator = new RegionSizeCalculator(regionLocator, admin);
258
259 Pair<byte[][], byte[][]> keys = getStartEndKeys();
260 if (keys == null || keys.getFirst() == null ||
261 keys.getFirst().length == 0) {
262 HRegionLocation regLoc = regionLocator.getRegionLocation(HConstants.EMPTY_BYTE_ARRAY, false);
263 if (null == regLoc) {
264 throw new IOException("Expecting at least one region.");
265 }
266 List<InputSplit> splits = new ArrayList<InputSplit>(1);
267 long regionSize = sizeCalculator.getRegionSize(regLoc.getRegionInfo().getRegionName());
268 TableSplit split = new TableSplit(table.getName(), scan,
269 HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, regLoc
270 .getHostnamePort().split(Addressing.HOSTNAME_PORT_SEPARATOR)[0], regionSize);
271 splits.add(split);
272 return splits;
273 }
274 List<InputSplit> splits = new ArrayList<InputSplit>(keys.getFirst().length);
275 for (int i = 0; i < keys.getFirst().length; i++) {
276 if ( !includeRegionInSplit(keys.getFirst()[i], keys.getSecond()[i])) {
277 continue;
278 }
279 HRegionLocation location = regionLocator.getRegionLocation(keys.getFirst()[i], false);
280
281 InetSocketAddress isa = new InetSocketAddress(location.getHostname(), location.getPort());
282 if (isa.isUnresolved()) {
283 LOG.warn("Failed resolve " + isa);
284 }
285 InetAddress regionAddress = isa.getAddress();
286 String regionLocation;
287 try {
288 regionLocation = reverseDNS(regionAddress);
289 } catch (NamingException e) {
290 LOG.warn("Cannot resolve the host name for " + regionAddress + " because of " + e);
291 regionLocation = location.getHostname();
292 }
293
294 byte[] startRow = scan.getStartRow();
295 byte[] stopRow = scan.getStopRow();
296
297 if ((startRow.length == 0 || keys.getSecond()[i].length == 0 ||
298 Bytes.compareTo(startRow, keys.getSecond()[i]) < 0) &&
299 (stopRow.length == 0 ||
300 Bytes.compareTo(stopRow, keys.getFirst()[i]) > 0)) {
301 byte[] splitStart = startRow.length == 0 ||
302 Bytes.compareTo(keys.getFirst()[i], startRow) >= 0 ?
303 keys.getFirst()[i] : startRow;
304 byte[] splitStop = (stopRow.length == 0 ||
305 Bytes.compareTo(keys.getSecond()[i], stopRow) <= 0) &&
306 keys.getSecond()[i].length > 0 ?
307 keys.getSecond()[i] : stopRow;
308
309 byte[] regionName = location.getRegionInfo().getRegionName();
310 String encodedRegionName = location.getRegionInfo().getEncodedName();
311 long regionSize = sizeCalculator.getRegionSize(regionName);
312 TableSplit split = new TableSplit(table.getName(), scan,
313 splitStart, splitStop, regionLocation, encodedRegionName, regionSize);
314 splits.add(split);
315 if (LOG.isDebugEnabled()) {
316 LOG.debug("getSplits: split -> " + i + " -> " + split);
317 }
318 }
319 }
320
321 boolean enableAutoBalance = context.getConfiguration().getBoolean(
322 MAPREDUCE_INPUT_AUTOBALANCE, false);
323 if (enableAutoBalance) {
324 long totalRegionSize=0;
325 for (int i = 0; i < splits.size(); i++){
326 TableSplit ts = (TableSplit)splits.get(i);
327 totalRegionSize += ts.getLength();
328 }
329 long averageRegionSize = totalRegionSize / splits.size();
330
331 if (averageRegionSize <= 0) {
332 LOG.warn("The averageRegionSize is not positive: "+ averageRegionSize + ", " +
333 "set it to 1.");
334 averageRegionSize = 1;
335 }
336 return calculateRebalancedSplits(splits, context, averageRegionSize);
337 } else {
338 return splits;
339 }
340 } finally {
341 if (closeOnFinish) {
342 closeTable();
343 }
344 }
345 }
346
347
348
349
350 @Deprecated
351 public String reverseDNS(InetAddress ipAddress) throws NamingException, UnknownHostException {
352 String hostName = this.reverseDNSCacheMap.get(ipAddress);
353 if (hostName == null) {
354 String ipAddressString = null;
355 try {
356 ipAddressString = DNS.reverseDns(ipAddress, null);
357 } catch (Exception e) {
358
359
360
361 ipAddressString = InetAddress.getByName(ipAddress.getHostAddress()).getHostName();
362 }
363 if (ipAddressString == null) throw new UnknownHostException("No host found for " + ipAddress);
364 hostName = Strings.domainNamePointerToHostName(ipAddressString);
365 this.reverseDNSCacheMap.put(ipAddress, hostName);
366 }
367 return hostName;
368 }
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383 public List<InputSplit> calculateRebalancedSplits(List<InputSplit> list, JobContext context,
384 long average) throws IOException {
385 List<InputSplit> resultList = new ArrayList<InputSplit>();
386 Configuration conf = context.getConfiguration();
387
388 long dataSkewRatio = conf.getLong(INPUT_AUTOBALANCE_MAXSKEWRATIO, 3);
389
390 boolean isTextKey = context.getConfiguration().getBoolean(TABLE_ROW_TEXTKEY, true);
391 long dataSkewThreshold = dataSkewRatio * average;
392 int count = 0;
393 while (count < list.size()) {
394 TableSplit ts = (TableSplit)list.get(count);
395 TableName tableName = ts.getTable();
396 String regionLocation = ts.getRegionLocation();
397 String encodedRegionName = ts.getEncodedRegionName();
398 long regionSize = ts.getLength();
399 if (regionSize >= dataSkewThreshold) {
400
401
402 byte[] splitKey = getSplitKey(ts.getStartRow(), ts.getEndRow(), isTextKey);
403 if (Arrays.equals(ts.getEndRow(), splitKey)) {
404
405 resultList.add(ts);
406 } else {
407
408
409 TableSplit t1 = new TableSplit(tableName, scan, ts.getStartRow(), splitKey,
410 regionLocation, regionSize / 2);
411 TableSplit t2 = new TableSplit(tableName, scan, splitKey, ts.getEndRow(), regionLocation,
412 regionSize - regionSize / 2);
413 resultList.add(t1);
414 resultList.add(t2);
415 }
416 count++;
417 } else if (regionSize >= average) {
418
419
420 resultList.add(ts);
421 count++;
422 } else {
423
424
425 long totalSize = regionSize;
426 byte[] splitStartKey = ts.getStartRow();
427 byte[] splitEndKey = ts.getEndRow();
428 count++;
429 for (; count < list.size(); count++) {
430 TableSplit nextRegion = (TableSplit)list.get(count);
431 long nextRegionSize = nextRegion.getLength();
432 if (totalSize + nextRegionSize <= dataSkewThreshold) {
433 totalSize = totalSize + nextRegionSize;
434 splitEndKey = nextRegion.getEndRow();
435 } else {
436 break;
437 }
438 }
439 TableSplit t = new TableSplit(table.getName(), scan, splitStartKey, splitEndKey,
440 regionLocation, encodedRegionName, totalSize);
441 resultList.add(t);
442 }
443 }
444 return resultList;
445 }
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492 public static byte[] getSplitKey(byte[] start, byte[] end, boolean isText) {
493 byte upperLimitByte;
494 byte lowerLimitByte;
495
496 if (isText) {
497
498
499 upperLimitByte = '~';
500 lowerLimitByte = ' ';
501 } else {
502 upperLimitByte = -1;
503 lowerLimitByte = 0;
504 }
505
506
507
508 if (start.length == 0 && end.length == 0){
509 return new byte[]{(byte) ((lowerLimitByte + upperLimitByte) / 2)};
510 }
511 if (start.length == 0 && end.length != 0){
512 return new byte[]{ end[0] };
513 }
514 if (start.length != 0 && end.length == 0){
515 byte[] result =new byte[start.length];
516 result[0]=start[0];
517 for (int k = 1; k < start.length; k++){
518 result[k] = upperLimitByte;
519 }
520 return result;
521 }
522 return Bytes.split(start, end, false, 1)[1];
523 }
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549 protected boolean includeRegionInSplit(final byte[] startKey, final byte [] endKey) {
550 return true;
551 }
552
553
554
555
556
557
558 @Deprecated
559 protected HTable getHTable() {
560 return (HTable) this.getTable();
561 }
562
563
564
565
566 protected RegionLocator getRegionLocator() {
567 if (regionLocator == null) {
568 throw new IllegalStateException(NOT_INITIALIZED);
569 }
570 return regionLocator;
571 }
572
573
574
575
576 protected Table getTable() {
577 if (table == null) {
578 throw new IllegalStateException(NOT_INITIALIZED);
579 }
580 return table;
581 }
582
583
584
585
586 protected Admin getAdmin() {
587 if (admin == null) {
588 throw new IllegalStateException(NOT_INITIALIZED);
589 }
590 return admin;
591 }
592
593
594
595
596
597
598
599
600
601
602
603 @Deprecated
604 protected void setHTable(HTable table) throws IOException {
605 this.table = table;
606 this.connection = table.getConnection();
607 try {
608 this.regionLocator = table.getRegionLocator();
609 this.admin = this.connection.getAdmin();
610 } catch (NeedUnmanagedConnectionException exception) {
611 LOG.warn("You are using an HTable instance that relies on an HBase-managed Connection. " +
612 "This is usually due to directly creating an HTable, which is deprecated. Instead, you " +
613 "should create a Connection object and then request a Table instance from it. If you " +
614 "don't need the Table instance for your own use, you should instead use the " +
615 "TableInputFormatBase.initalizeTable method directly.");
616 LOG.info("Creating an additional unmanaged connection because user provided one can't be " +
617 "used for administrative actions. We'll close it when we close out the table.");
618 LOG.debug("Details about our failure to request an administrative interface.", exception);
619
620
621 this.connection = ConnectionFactory.createConnection(this.connection.getConfiguration());
622 this.regionLocator = this.connection.getRegionLocator(table.getName());
623 this.admin = this.connection.getAdmin();
624 }
625 }
626
627
628
629
630
631
632
633
634 protected void initializeTable(Connection connection, TableName tableName) throws IOException {
635 if (this.table != null || this.connection != null) {
636 LOG.warn("initializeTable called multiple times. Overwriting connection and table " +
637 "reference; TableInputFormatBase will not close these old references when done.");
638 }
639 this.table = connection.getTable(tableName);
640 this.regionLocator = connection.getRegionLocator(tableName);
641 this.admin = connection.getAdmin();
642 this.connection = connection;
643 }
644
645
646
647
648
649
650 public Scan getScan() {
651 if (this.scan == null) this.scan = new Scan();
652 return scan;
653 }
654
655
656
657
658
659
660 public void setScan(Scan scan) {
661 this.scan = scan;
662 }
663
664
665
666
667
668
669
670 protected void setTableRecordReader(TableRecordReader tableRecordReader) {
671 this.tableRecordReader = tableRecordReader;
672 }
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689 protected void initialize(JobContext context) throws IOException {
690 }
691
692
693
694
695
696
697
698 protected void closeTable() throws IOException {
699 close(admin, table, regionLocator, connection);
700 admin = null;
701 table = null;
702 regionLocator = null;
703 connection = null;
704 }
705
706 private void close(Closeable... closables) throws IOException {
707 for (Closeable c : closables) {
708 if(c != null) { c.close(); }
709 }
710 }
711
712 }