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