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 static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertTrue;
24
25 import java.io.IOException;
26 import java.util.HashMap;
27 import java.util.HashSet;
28 import java.util.Iterator;
29 import java.util.List;
30 import java.util.Map;
31 import java.util.Set;
32 import java.util.UUID;
33
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.conf.Configurable;
37 import org.apache.hadoop.conf.Configuration;
38 import org.apache.hadoop.fs.FSDataOutputStream;
39 import org.apache.hadoop.fs.FileStatus;
40 import org.apache.hadoop.fs.FileSystem;
41 import org.apache.hadoop.fs.Path;
42 import org.apache.hadoop.hbase.Cell;
43 import org.apache.hadoop.hbase.CellUtil;
44 import org.apache.hadoop.hbase.HBaseConfiguration;
45 import org.apache.hadoop.hbase.HBaseTestingUtility;
46 import org.apache.hadoop.hbase.HConstants;
47 import org.apache.hadoop.hbase.testclassification.LargeTests;
48 import org.apache.hadoop.hbase.TableName;
49 import org.apache.hadoop.hbase.TableNotFoundException;
50 import org.apache.hadoop.hbase.client.HTable;
51 import org.apache.hadoop.hbase.client.Result;
52 import org.apache.hadoop.hbase.client.ResultScanner;
53 import org.apache.hadoop.hbase.client.Scan;
54 import org.apache.hadoop.hbase.client.Table;
55 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
56 import org.apache.hadoop.hbase.io.hfile.HFile;
57 import org.apache.hadoop.hbase.io.hfile.HFileScanner;
58 import org.apache.hadoop.hbase.util.Bytes;
59 import org.apache.hadoop.io.Text;
60 import org.apache.hadoop.mapred.Utils.OutputFileUtils.OutputFilesFilter;
61 import org.apache.hadoop.mapreduce.Job;
62 import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
63 import org.apache.hadoop.util.GenericOptionsParser;
64 import org.apache.hadoop.util.Tool;
65 import org.apache.hadoop.util.ToolRunner;
66 import org.junit.AfterClass;
67 import org.junit.Before;
68 import org.junit.BeforeClass;
69 import org.junit.Rule;
70 import org.junit.Test;
71 import org.junit.experimental.categories.Category;
72 import org.junit.rules.ExpectedException;
73
74 @Category(LargeTests.class)
75 public class TestImportTsv implements Configurable {
76
77 private static final Log LOG = LogFactory.getLog(TestImportTsv.class);
78 protected static final String NAME = TestImportTsv.class.getSimpleName();
79 protected static HBaseTestingUtility util = new HBaseTestingUtility();
80
81
82 protected static final String DELETE_AFTER_LOAD_CONF = NAME + ".deleteAfterLoad";
83
84
85
86
87 protected static final String FORCE_COMBINER_CONF = NAME + ".forceCombiner";
88
89 private final String FAMILY = "FAM";
90 private String table;
91 private Map<String, String> args;
92
93 @Rule
94 public ExpectedException exception = ExpectedException.none();
95
96 public Configuration getConf() {
97 return util.getConfiguration();
98 }
99
100 public void setConf(Configuration conf) {
101 throw new IllegalArgumentException("setConf not supported");
102 }
103
104 @BeforeClass
105 public static void provisionCluster() throws Exception {
106 util.setJobWithoutMRCluster();
107 util.startMiniCluster();
108 }
109
110 @AfterClass
111 public static void releaseCluster() throws Exception {
112 util.shutdownMiniCluster();
113 }
114
115 @Before
116 public void setup() throws Exception {
117 table = "test-" + UUID.randomUUID();
118 args = new HashMap<String, String>();
119
120 args.put(ImportTsv.COLUMNS_CONF_KEY, "HBASE_ROW_KEY,FAM:A,FAM:B");
121 args.put(ImportTsv.SEPARATOR_CONF_KEY, "\u001b");
122 }
123
124 @Test
125 public void testMROnTable() throws Exception {
126 util.createTable(TableName.valueOf(table), FAMILY);
127 doMROnTableTest(null, 1);
128 util.deleteTable(table);
129 }
130
131 @Test
132 public void testMROnTableWithTimestamp() throws Exception {
133 util.createTable(TableName.valueOf(table), FAMILY);
134 args.put(ImportTsv.COLUMNS_CONF_KEY, "HBASE_ROW_KEY,HBASE_TS_KEY,FAM:A,FAM:B");
135 args.put(ImportTsv.SEPARATOR_CONF_KEY, ",");
136 String data = "KEY,1234,VALUE1,VALUE2\n";
137
138 doMROnTableTest(data, 1);
139 util.deleteTable(table);
140 }
141
142 @Test
143 public void testMROnTableWithCustomMapper()
144 throws Exception {
145 util.createTable(TableName.valueOf(table), FAMILY);
146 args.put(ImportTsv.MAPPER_CONF_KEY,
147 "org.apache.hadoop.hbase.mapreduce.TsvImporterCustomTestMapper");
148
149 doMROnTableTest(null, 3);
150 util.deleteTable(table);
151 }
152
153 @Test
154 public void testBulkOutputWithoutAnExistingTable() throws Exception {
155
156 Path hfiles = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
157 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, hfiles.toString());
158
159 doMROnTableTest(null, 3);
160 util.deleteTable(table);
161 }
162
163 @Test
164 public void testBulkOutputWithAnExistingTable() throws Exception {
165 util.createTable(TableName.valueOf(table), FAMILY);
166
167
168 Path hfiles = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
169 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, hfiles.toString());
170
171 doMROnTableTest(null, 3);
172 util.deleteTable(table);
173 }
174
175 @Test
176 public void testBulkOutputWithAnExistingTableNoStrictTrue() throws Exception {
177 util.createTable(TableName.valueOf(table), FAMILY);
178
179
180 Path hfiles = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
181 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, hfiles.toString());
182 args.put(ImportTsv.NO_STRICT_COL_FAMILY, "true");
183 doMROnTableTest(null, 3);
184 util.deleteTable(table);
185 }
186
187 @Test
188 public void testJobConfigurationsWithTsvImporterTextMapper() throws Exception {
189 Path bulkOutputPath = new Path(util.getDataTestDirOnTestFS(table),"hfiles");
190 String INPUT_FILE = "InputFile1.csv";
191
192 String[] args =
193 new String[] {
194 "-D" + ImportTsv.MAPPER_CONF_KEY
195 + "=org.apache.hadoop.hbase.mapreduce.TsvImporterTextMapper",
196 "-D" + ImportTsv.COLUMNS_CONF_KEY
197 + "=HBASE_ROW_KEY,FAM:A,FAM:B",
198 "-D" + ImportTsv.SEPARATOR_CONF_KEY + "=,",
199 "-D" + ImportTsv.BULK_OUTPUT_CONF_KEY + "=" + bulkOutputPath.toString(), table,
200 INPUT_FILE
201 };
202 Configuration conf = new Configuration(util.getConfiguration());
203 GenericOptionsParser opts = new GenericOptionsParser(conf, args);
204 args = opts.getRemainingArgs();
205 assertEquals("running test job configuration failed.", 0,
206 ToolRunner.run(conf, new ImportTsv() {
207 @Override
208 public int run(String[] args) throws Exception {
209 Job job = createSubmittableJob(getConf(), args);
210 assertTrue(job.getMapperClass().equals(TsvImporterTextMapper.class));
211 assertTrue(job.getReducerClass().equals(TextSortReducer.class));
212 assertTrue(job.getMapOutputValueClass().equals(Text.class));
213 return 0;
214 }
215 }, args));
216
217 util.deleteTable(table);
218 }
219
220 @Test
221 public void testBulkOutputWithTsvImporterTextMapper() throws Exception {
222 Path bulkOutputPath = new Path(util.getDataTestDirOnTestFS(table),"hfiles");
223 args.put(ImportTsv.MAPPER_CONF_KEY, "org.apache.hadoop.hbase.mapreduce.TsvImporterTextMapper");
224 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, bulkOutputPath.toString());
225 String data = "KEY\u001bVALUE4\u001bVALUE8\n";
226 doMROnTableTest(data, 4);
227 util.deleteTable(table);
228 }
229
230 @Test
231 public void testWithoutAnExistingTableAndCreateTableSetToNo() throws Exception {
232 String[] args = new String[] { table, "/inputFile" };
233
234 Configuration conf = new Configuration(util.getConfiguration());
235 conf.set(ImportTsv.COLUMNS_CONF_KEY, "HBASE_ROW_KEY,FAM:A");
236 conf.set(ImportTsv.BULK_OUTPUT_CONF_KEY, "/output");
237 conf.set(ImportTsv.CREATE_TABLE_CONF_KEY, "no");
238 exception.expect(TableNotFoundException.class);
239 assertEquals("running test job configuration failed.", 0,
240 ToolRunner.run(conf, new ImportTsv() {
241 @Override public int run(String[] args) throws Exception {
242 createSubmittableJob(getConf(), args);
243 return 0;
244 }
245 }, args));
246 }
247
248 @Test
249 public void testMRWithoutAnExistingTable() throws Exception {
250 String[] args =
251 new String[] { table, "/inputFile" };
252
253 exception.expect(TableNotFoundException.class);
254 assertEquals("running test job configuration failed.", 0, ToolRunner.run(
255 new Configuration(util.getConfiguration()),
256 new ImportTsv() {
257 @Override
258 public int run(String[] args) throws Exception {
259 createSubmittableJob(getConf(), args);
260 return 0;
261 }
262 }, args));
263 }
264
265 @Test
266 public void testJobConfigurationsWithDryMode() throws Exception {
267 Path bulkOutputPath = new Path(util.getDataTestDirOnTestFS(table),"hfiles");
268 String INPUT_FILE = "InputFile1.csv";
269
270 String[] argsArray = new String[] {
271 "-D" + ImportTsv.COLUMNS_CONF_KEY + "=HBASE_ROW_KEY,FAM:A,FAM:B",
272 "-D" + ImportTsv.SEPARATOR_CONF_KEY + "=,",
273 "-D" + ImportTsv.BULK_OUTPUT_CONF_KEY + "=" + bulkOutputPath.toString(),
274 "-D" + ImportTsv.DRY_RUN_CONF_KEY + "=true",
275 table,
276 INPUT_FILE };
277 assertEquals("running test job configuration failed.", 0, ToolRunner.run(
278 new Configuration(util.getConfiguration()),
279 new ImportTsv() {
280 @Override
281 public int run(String[] args) throws Exception {
282 Job job = createSubmittableJob(getConf(), args);
283 assertTrue(job.getOutputFormatClass().equals(NullOutputFormat.class));
284 return 0;
285 }
286 }, argsArray));
287
288 util.deleteTable(table);
289 }
290
291 @Test
292 public void testDryModeWithoutBulkOutputAndTableExists() throws Exception {
293 util.createTable(TableName.valueOf(table), FAMILY);
294 args.put(ImportTsv.DRY_RUN_CONF_KEY, "true");
295 doMROnTableTest(null, 1);
296
297
298 util.deleteTable(table);
299 }
300
301
302
303
304
305 @Test
306 public void testDryModeWithoutBulkOutputAndTableDoesNotExists() throws Exception {
307 args.put(ImportTsv.DRY_RUN_CONF_KEY, "true");
308 exception.expect(TableNotFoundException.class);
309 doMROnTableTest(null, 1);
310 }
311
312 @Test public void testDryModeWithBulkOutputAndTableExists() throws Exception {
313 util.createTable(TableName.valueOf(table), FAMILY);
314
315 Path hfiles = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
316 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, hfiles.toString());
317 args.put(ImportTsv.DRY_RUN_CONF_KEY, "true");
318 doMROnTableTest(null, 1);
319
320
321 util.deleteTable(table);
322 }
323
324
325
326
327
328 @Test
329 public void testDryModeWithBulkOutputAndTableDoesNotExistsCreateTableSetToNo() throws
330 Exception {
331
332 Path hfiles = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
333 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, hfiles.toString());
334 args.put(ImportTsv.DRY_RUN_CONF_KEY, "true");
335 args.put(ImportTsv.CREATE_TABLE_CONF_KEY, "no");
336 exception.expect(TableNotFoundException.class);
337 doMROnTableTest(null, 1);
338 }
339
340 @Test
341 public void testDryModeWithBulkModeAndTableDoesNotExistsCreateTableSetToYes() throws Exception {
342
343 Path hfiles = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
344 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, hfiles.toString());
345 args.put(ImportTsv.DRY_RUN_CONF_KEY, "true");
346 args.put(ImportTsv.CREATE_TABLE_CONF_KEY, "yes");
347 doMROnTableTest(null, 1);
348
349 exception.expect(TableNotFoundException.class);
350 util.deleteTable(table);
351 }
352
353
354
355
356 @Test
357 public void testTsvImporterTextMapperWithInvalidData() throws Exception {
358 Path bulkOutputPath = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
359 args.put(ImportTsv.MAPPER_CONF_KEY, "org.apache.hadoop.hbase.mapreduce.TsvImporterTextMapper");
360 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, bulkOutputPath.toString());
361 args.put(ImportTsv.COLUMNS_CONF_KEY, "HBASE_ROW_KEY,HBASE_TS_KEY,FAM:A,FAM:B");
362 args.put(ImportTsv.SEPARATOR_CONF_KEY, ",");
363
364 String data = "KEY,1234,VALUE1,VALUE2\nKEY\nKEY,1235,VALUE1,VALUE2\n";
365 doMROnTableTest(data, 1, 4);
366 util.deleteTable(table);
367 }
368
369 @Test
370 public void testSkipEmptyColumns() throws Exception {
371 Path bulkOutputPath = new Path(util.getDataTestDirOnTestFS(table), "hfiles");
372 args.put(ImportTsv.BULK_OUTPUT_CONF_KEY, bulkOutputPath.toString());
373 args.put(ImportTsv.COLUMNS_CONF_KEY, "HBASE_ROW_KEY,HBASE_TS_KEY,FAM:A,FAM:B");
374 args.put(ImportTsv.SEPARATOR_CONF_KEY, ",");
375 args.put(ImportTsv.SKIP_EMPTY_COLUMNS, "true");
376
377 String data = "KEY,1234,VALUE1,VALUE2\nKEY,1235,,VALUE2\n";
378 doMROnTableTest(util, table, FAMILY, data, args, 1, 3);
379 util.deleteTable(table);
380 }
381
382 private Tool doMROnTableTest(String data, int valueMultiplier,int expectedKVCount)
383 throws Exception {
384 return doMROnTableTest(util, table, FAMILY, data, args, valueMultiplier,expectedKVCount);
385 }
386
387 private Tool doMROnTableTest(String data, int valueMultiplier) throws Exception {
388 return doMROnTableTest(util, table, FAMILY, data, args, valueMultiplier,-1);
389 }
390
391 protected static Tool doMROnTableTest(HBaseTestingUtility util, String table,
392 String family, String data, Map<String, String> args) throws Exception {
393 return doMROnTableTest(util, table, family, data, args, 1,-1);
394 }
395
396
397
398
399
400
401
402
403
404 protected static Tool doMROnTableTest(HBaseTestingUtility util, String table,
405 String family, String data, Map<String, String> args, int valueMultiplier,int expectedKVCount)
406 throws Exception {
407 Configuration conf = new Configuration(util.getConfiguration());
408
409
410 FileSystem fs = FileSystem.get(conf);
411 Path inputPath = fs.makeQualified(new Path(util.getDataTestDirOnTestFS(table), "input.dat"));
412 FSDataOutputStream op = fs.create(inputPath, true);
413 if (data == null) {
414 data = "KEY\u001bVALUE1\u001bVALUE2\n";
415 }
416 op.write(Bytes.toBytes(data));
417 op.close();
418 LOG.debug(String.format("Wrote test data to file: %s", inputPath));
419
420 if (conf.getBoolean(FORCE_COMBINER_CONF, true)) {
421 LOG.debug("Forcing combiner.");
422 conf.setInt("mapreduce.map.combine.minspills", 1);
423 }
424
425
426 String[] argsArray = new String[args.size() + 2];
427 Iterator it = args.entrySet().iterator();
428 int i = 0;
429 while (it.hasNext()) {
430 Map.Entry pair = (Map.Entry) it.next();
431 argsArray[i] = "-D" + pair.getKey() + "=" + pair.getValue();
432 i++;
433 }
434 argsArray[i] = table;
435 argsArray[i + 1] = inputPath.toString();
436
437
438 Tool tool = new ImportTsv();
439 LOG.debug("Running ImportTsv with arguments: " + argsArray);
440 assertEquals(0, ToolRunner.run(conf, tool, argsArray));
441
442
443
444
445 boolean isDryRun = args.containsKey(ImportTsv.DRY_RUN_CONF_KEY) &&
446 "true".equalsIgnoreCase(args.get(ImportTsv.DRY_RUN_CONF_KEY));
447 if (args.containsKey(ImportTsv.BULK_OUTPUT_CONF_KEY)) {
448 if (isDryRun) {
449 assertFalse(String.format("Dry run mode, %s should not have been created.",
450 ImportTsv.BULK_OUTPUT_CONF_KEY),
451 fs.exists(new Path(ImportTsv.BULK_OUTPUT_CONF_KEY)));
452 } else {
453 validateHFiles(fs, args.get(ImportTsv.BULK_OUTPUT_CONF_KEY), family,expectedKVCount);
454 }
455 } else {
456 validateTable(conf, TableName.valueOf(table), family, valueMultiplier, isDryRun);
457 }
458
459 if (conf.getBoolean(DELETE_AFTER_LOAD_CONF, true)) {
460 LOG.debug("Deleting test subdirectory");
461 util.cleanupDataTestDirOnTestFS(table);
462 }
463 return tool;
464 }
465
466
467
468
469 private static void validateTable(Configuration conf, TableName tableName,
470 String family, int valueMultiplier, boolean isDryRun) throws IOException {
471
472 LOG.debug("Validating table.");
473 Table table = new HTable(conf, tableName);
474 boolean verified = false;
475 long pause = conf.getLong("hbase.client.pause", 5 * 1000);
476 int numRetries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 5);
477 for (int i = 0; i < numRetries; i++) {
478 try {
479 Scan scan = new Scan();
480
481 scan.addFamily(Bytes.toBytes(family));
482 ResultScanner resScanner = table.getScanner(scan);
483 int numRows = 0;
484 for (Result res : resScanner) {
485 numRows++;
486 assertEquals(2, res.size());
487 List<Cell> kvs = res.listCells();
488 assertTrue(CellUtil.matchingRow(kvs.get(0), Bytes.toBytes("KEY")));
489 assertTrue(CellUtil.matchingRow(kvs.get(1), Bytes.toBytes("KEY")));
490 assertTrue(CellUtil.matchingValue(kvs.get(0), Bytes.toBytes("VALUE" + valueMultiplier)));
491 assertTrue(CellUtil.matchingValue(kvs.get(1), Bytes.toBytes("VALUE" + 2 * valueMultiplier)));
492
493 }
494 if (isDryRun) {
495 assertEquals(0, numRows);
496 } else {
497 assertEquals(1, numRows);
498 }
499 verified = true;
500 break;
501 } catch (NullPointerException e) {
502
503
504 }
505 try {
506 Thread.sleep(pause);
507 } catch (InterruptedException e) {
508
509 }
510 }
511 table.close();
512 assertTrue(verified);
513 }
514
515
516
517
518 private static void validateHFiles(FileSystem fs, String outputPath, String family,
519 int expectedKVCount) throws IOException {
520
521 LOG.debug("Validating HFiles.");
522 Set<String> configFamilies = new HashSet<String>();
523 configFamilies.add(family);
524 Set<String> foundFamilies = new HashSet<String>();
525 int actualKVCount = 0;
526 for (FileStatus cfStatus : fs.listStatus(new Path(outputPath), new OutputFilesFilter())) {
527 String[] elements = cfStatus.getPath().toString().split(Path.SEPARATOR);
528 String cf = elements[elements.length - 1];
529 foundFamilies.add(cf);
530 assertTrue(
531 String.format(
532 "HFile output contains a column family (%s) not present in input families (%s)",
533 cf, configFamilies),
534 configFamilies.contains(cf));
535 for (FileStatus hfile : fs.listStatus(cfStatus.getPath())) {
536 assertTrue(
537 String.format("HFile %s appears to contain no data.", hfile.getPath()),
538 hfile.getLen() > 0);
539
540 if (expectedKVCount > -1) {
541 actualKVCount += getKVCountFromHfile(fs, hfile.getPath());
542 }
543 }
544 }
545 assertTrue(String.format("HFile output does not contain the input family '%s'.", family),
546 foundFamilies.contains(family));
547 if (expectedKVCount > -1) {
548 assertTrue(String.format(
549 "KV count in ouput hfile=<%d> doesn't match with expected KV count=<%d>", actualKVCount,
550 expectedKVCount), actualKVCount == expectedKVCount);
551 }
552 }
553
554
555
556
557
558
559
560
561 private static int getKVCountFromHfile(FileSystem fs, Path p) throws IOException {
562 Configuration conf = util.getConfiguration();
563 HFile.Reader reader = HFile.createReader(fs, p, new CacheConfig(conf), conf);
564 reader.loadFileInfo();
565 HFileScanner scanner = reader.getScanner(false, false);
566 scanner.seekTo();
567 int count = 0;
568 do {
569 count++;
570 } while (scanner.next());
571 reader.close();
572 return count;
573 }
574 }
575