| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.hcatalog.hbase; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.hbase.HBaseConfiguration; |
| import org.apache.hadoop.hbase.client.HTable; |
| import org.apache.hadoop.hbase.client.Put; |
| import org.apache.hadoop.hbase.client.Result; |
| import org.apache.hadoop.hbase.client.ResultScanner; |
| import org.apache.hadoop.hbase.client.Scan; |
| import org.apache.hadoop.hbase.io.ImmutableBytesWritable; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hive.cli.CliSessionState; |
| import org.apache.hadoop.hive.conf.HiveConf; |
| import org.apache.hadoop.hive.ql.session.SessionState; |
| import org.apache.hadoop.io.BytesWritable; |
| import org.apache.hadoop.io.LongWritable; |
| import org.apache.hadoop.io.Text; |
| import org.apache.hadoop.mapred.JobClient; |
| import org.apache.hadoop.mapred.JobConf; |
| import org.apache.hadoop.mapred.OutputCollector; |
| import org.apache.hadoop.mapred.Reporter; |
| import org.apache.hadoop.mapred.RunningJob; |
| import org.apache.hadoop.mapreduce.Job; |
| import org.apache.hadoop.mapreduce.Mapper; |
| import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; |
| import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; |
| import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; |
| import org.apache.hcatalog.cli.HCatDriver; |
| import org.apache.hcatalog.cli.SemanticAnalysis.HCatSemanticAnalyzer; |
| import org.apache.hcatalog.common.HCatConstants; |
| import org.apache.hcatalog.common.HCatUtil; |
| import org.apache.hcatalog.data.DefaultHCatRecord; |
| import org.apache.hcatalog.data.HCatRecord; |
| import org.apache.hcatalog.data.schema.HCatSchema; |
| import org.apache.hcatalog.hbase.HBaseBulkOutputFormat.HBaseBulkOutputCommitter; |
| import org.apache.hcatalog.hbase.TestHBaseDirectOutputFormat.MapReadAbortedTransaction; |
| import org.apache.hcatalog.hbase.TestHBaseDirectOutputFormat.MapWriteAbortTransaction; |
| import org.apache.hcatalog.hbase.snapshot.FamilyRevision; |
| import org.apache.hcatalog.hbase.snapshot.RevisionManager; |
| import org.apache.hcatalog.hbase.snapshot.RevisionManagerConfiguration; |
| import org.apache.hcatalog.hbase.snapshot.TableSnapshot; |
| import org.apache.hcatalog.hbase.snapshot.Transaction; |
| import org.apache.hcatalog.mapreduce.HCatInputFormat; |
| import org.apache.hcatalog.mapreduce.HCatOutputFormat; |
| import org.apache.hcatalog.mapreduce.InputJobInfo; |
| import org.apache.hcatalog.mapreduce.OutputJobInfo; |
| |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.Map; |
| |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| |
| /** |
| * Tests components of HBaseHCatStorageHandler using ManyMiniCluster. |
| * Including ImprtSequenceFile and HBaseBulkOutputFormat |
| */ |
| public class TestHBaseBulkOutputFormat extends SkeletonHBaseTest { |
| private final static Logger LOG = LoggerFactory.getLogger(TestHBaseBulkOutputFormat.class); |
| |
| private final HiveConf allConf; |
| private final HCatDriver hcatDriver; |
| |
| public TestHBaseBulkOutputFormat() { |
| allConf = getHiveConf(); |
| allConf.set(HiveConf.ConfVars.SEMANTIC_ANALYZER_HOOK.varname, |
| HCatSemanticAnalyzer.class.getName()); |
| allConf.set(HiveConf.ConfVars.HADOOPFS.varname, getFileSystem().getUri().toString()); |
| allConf.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname, new Path(getTestDir(), "warehouse").toString()); |
| |
| //Add hbase properties |
| for (Map.Entry<String, String> el : getHbaseConf()) |
| allConf.set(el.getKey(), el.getValue()); |
| for (Map.Entry<String, String> el : getJobConf()) |
| allConf.set(el.getKey(), el.getValue()); |
| |
| HBaseConfiguration.merge( |
| allConf, |
| RevisionManagerConfiguration.create()); |
| SessionState.start(new CliSessionState(allConf)); |
| hcatDriver = new HCatDriver(); |
| } |
| |
| public static class MapWriteOldMapper implements org.apache.hadoop.mapred.Mapper<LongWritable, Text, ImmutableBytesWritable, Put> { |
| |
| @Override |
| public void close() throws IOException { |
| } |
| |
| @Override |
| public void configure(JobConf job) { |
| } |
| |
| @Override |
| public void map(LongWritable key, Text value, |
| OutputCollector<ImmutableBytesWritable, Put> output, |
| Reporter reporter) throws IOException { |
| String vals[] = value.toString().split(","); |
| Put put = new Put(Bytes.toBytes(vals[0])); |
| for (int i = 1; i < vals.length; i++) { |
| String pair[] = vals[i].split(":"); |
| put.add(Bytes.toBytes("my_family"), |
| Bytes.toBytes(pair[0]), |
| Bytes.toBytes(pair[1])); |
| } |
| output.collect(new ImmutableBytesWritable(Bytes.toBytes(vals[0])), put); |
| } |
| |
| } |
| |
| public static class MapWrite extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> { |
| |
| @Override |
| public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { |
| String vals[] = value.toString().split(","); |
| Put put = new Put(Bytes.toBytes(vals[0])); |
| for (int i = 1; i < vals.length; i++) { |
| String pair[] = vals[i].split(":"); |
| put.add(Bytes.toBytes("my_family"), |
| Bytes.toBytes(pair[0]), |
| Bytes.toBytes(pair[1])); |
| } |
| context.write(new ImmutableBytesWritable(Bytes.toBytes(vals[0])), put); |
| } |
| } |
| |
| public static class MapHCatWrite extends Mapper<LongWritable, Text, BytesWritable, HCatRecord> { |
| @Override |
| public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { |
| OutputJobInfo jobInfo = (OutputJobInfo) HCatUtil.deserialize(context.getConfiguration().get(HCatConstants.HCAT_KEY_OUTPUT_INFO)); |
| HCatRecord record = new DefaultHCatRecord(3); |
| HCatSchema schema = jobInfo.getOutputSchema(); |
| String vals[] = value.toString().split(","); |
| record.setInteger("key", schema, Integer.parseInt(vals[0])); |
| for (int i = 1; i < vals.length; i++) { |
| String pair[] = vals[i].split(":"); |
| record.set(pair[0], schema, pair[1]); |
| } |
| context.write(null, record); |
| } |
| } |
| |
| @Test |
| public void hbaseBulkOutputFormatTest() throws IOException, ClassNotFoundException, InterruptedException { |
| String testName = "hbaseBulkOutputFormatTest"; |
| Path methodTestDir = new Path(getTestDir(), testName); |
| LOG.info("starting: " + testName); |
| |
| String tableName = newTableName(testName).toLowerCase(); |
| String familyName = "my_family"; |
| byte[] familyNameBytes = Bytes.toBytes(familyName); |
| |
| //include hbase config in conf file |
| Configuration conf = new Configuration(allConf); |
| |
| //create table |
| conf.set(HBaseConstants.PROPERTY_OUTPUT_TABLE_NAME_KEY, tableName); |
| createTable(tableName, new String[]{familyName}); |
| |
| String data[] = {"1,english:one,spanish:uno", |
| "2,english:two,spanish:dos", |
| "3,english:three,spanish:tres"}; |
| |
| |
| // input/output settings |
| Path inputPath = new Path(methodTestDir, "mr_input"); |
| FSDataOutputStream os = getFileSystem().create(new Path(inputPath, "inputFile.txt")); |
| for (String line : data) |
| os.write(Bytes.toBytes(line + "\n")); |
| os.close(); |
| Path interPath = new Path(methodTestDir, "inter"); |
| //create job |
| JobConf job = new JobConf(conf); |
| job.setWorkingDirectory(new Path(methodTestDir, "mr_work")); |
| job.setJarByClass(this.getClass()); |
| job.setMapperClass(MapWriteOldMapper.class); |
| |
| job.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class); |
| org.apache.hadoop.mapred.TextInputFormat.setInputPaths(job, inputPath); |
| |
| job.setOutputFormat(HBaseBulkOutputFormat.class); |
| org.apache.hadoop.mapred.SequenceFileOutputFormat.setOutputPath(job, interPath); |
| job.setOutputCommitter(HBaseBulkOutputCommitter.class); |
| |
| //manually create transaction |
| RevisionManager rm = HBaseRevisionManagerUtil.getOpenedRevisionManager(conf); |
| try { |
| OutputJobInfo outputJobInfo = OutputJobInfo.create("default", tableName, null); |
| Transaction txn = rm.beginWriteTransaction(tableName, Arrays.asList(familyName)); |
| outputJobInfo.getProperties().setProperty(HBaseConstants.PROPERTY_WRITE_TXN_KEY, |
| HCatUtil.serialize(txn)); |
| job.set(HCatConstants.HCAT_KEY_OUTPUT_INFO, |
| HCatUtil.serialize(outputJobInfo)); |
| } finally { |
| rm.close(); |
| } |
| |
| job.setMapOutputKeyClass(ImmutableBytesWritable.class); |
| job.setMapOutputValueClass(HCatRecord.class); |
| |
| job.setOutputKeyClass(ImmutableBytesWritable.class); |
| job.setOutputValueClass(HCatRecord.class); |
| |
| job.setNumReduceTasks(0); |
| |
| RunningJob runJob = JobClient.runJob(job); |
| runJob.waitForCompletion(); |
| assertTrue(runJob.isSuccessful()); |
| |
| //verify |
| HTable table = new HTable(conf, tableName); |
| Scan scan = new Scan(); |
| scan.addFamily(familyNameBytes); |
| ResultScanner scanner = table.getScanner(scan); |
| int index = 0; |
| for (Result result : scanner) { |
| String vals[] = data[index].toString().split(","); |
| for (int i = 1; i < vals.length; i++) { |
| String pair[] = vals[i].split(":"); |
| assertTrue(result.containsColumn(familyNameBytes, Bytes.toBytes(pair[0]))); |
| assertEquals(pair[1], Bytes.toString(result.getValue(familyNameBytes, Bytes.toBytes(pair[0])))); |
| } |
| index++; |
| } |
| //test if load count is the same |
| assertEquals(data.length, index); |
| //test if scratch directory was erased |
| assertFalse(FileSystem.get(job).exists(interPath)); |
| } |
| |
| @Test |
| public void importSequenceFileTest() throws IOException, ClassNotFoundException, InterruptedException { |
| String testName = "importSequenceFileTest"; |
| Path methodTestDir = new Path(getTestDir(), testName); |
| LOG.info("starting: " + testName); |
| |
| String tableName = newTableName(testName).toLowerCase(); |
| String familyName = "my_family"; |
| byte[] familyNameBytes = Bytes.toBytes(familyName); |
| |
| //include hbase config in conf file |
| Configuration conf = new Configuration(allConf); |
| |
| //create table |
| createTable(tableName, new String[]{familyName}); |
| |
| String data[] = {"1,english:one,spanish:uno", |
| "2,english:two,spanish:dos", |
| "3,english:three,spanish:tres"}; |
| |
| |
| // input/output settings |
| Path inputPath = new Path(methodTestDir, "mr_input"); |
| getFileSystem().mkdirs(inputPath); |
| FSDataOutputStream os = getFileSystem().create(new Path(inputPath, "inputFile.txt")); |
| for (String line : data) |
| os.write(Bytes.toBytes(line + "\n")); |
| os.close(); |
| Path interPath = new Path(methodTestDir, "inter"); |
| Path scratchPath = new Path(methodTestDir, "scratch"); |
| |
| |
| //create job |
| Job job = new Job(conf, testName); |
| job.setWorkingDirectory(new Path(methodTestDir, "mr_work")); |
| job.setJarByClass(this.getClass()); |
| job.setMapperClass(MapWrite.class); |
| |
| job.setInputFormatClass(TextInputFormat.class); |
| TextInputFormat.setInputPaths(job, inputPath); |
| |
| job.setOutputFormatClass(SequenceFileOutputFormat.class); |
| SequenceFileOutputFormat.setOutputPath(job, interPath); |
| |
| job.setMapOutputKeyClass(ImmutableBytesWritable.class); |
| job.setMapOutputValueClass(Put.class); |
| |
| job.setOutputKeyClass(ImmutableBytesWritable.class); |
| job.setOutputValueClass(Put.class); |
| |
| job.setNumReduceTasks(0); |
| assertTrue(job.waitForCompletion(true)); |
| |
| job = new Job(new Configuration(allConf), testName + "_importer"); |
| assertTrue(ImportSequenceFile.runJob(job, tableName, interPath, scratchPath)); |
| |
| //verify |
| HTable table = new HTable(conf, tableName); |
| Scan scan = new Scan(); |
| scan.addFamily(familyNameBytes); |
| ResultScanner scanner = table.getScanner(scan); |
| int index = 0; |
| for (Result result : scanner) { |
| String vals[] = data[index].toString().split(","); |
| for (int i = 1; i < vals.length; i++) { |
| String pair[] = vals[i].split(":"); |
| assertTrue(result.containsColumn(familyNameBytes, Bytes.toBytes(pair[0]))); |
| assertEquals(pair[1], Bytes.toString(result.getValue(familyNameBytes, Bytes.toBytes(pair[0])))); |
| } |
| index++; |
| } |
| //test if load count is the same |
| assertEquals(data.length, index); |
| //test if scratch directory was erased |
| assertFalse(FileSystem.get(job.getConfiguration()).exists(scratchPath)); |
| } |
| |
| @Test |
| public void bulkModeHCatOutputFormatTest() throws Exception { |
| String testName = "bulkModeHCatOutputFormatTest"; |
| Path methodTestDir = new Path(getTestDir(), testName); |
| LOG.info("starting: " + testName); |
| |
| String databaseName = testName.toLowerCase(); |
| String dbDir = new Path(methodTestDir, "DB_" + testName).toString(); |
| String tableName = newTableName(testName).toLowerCase(); |
| String familyName = "my_family"; |
| byte[] familyNameBytes = Bytes.toBytes(familyName); |
| |
| |
| //include hbase config in conf file |
| Configuration conf = new Configuration(allConf); |
| conf.set(HCatConstants.HCAT_KEY_HIVE_CONF, HCatUtil.serialize(allConf.getAllProperties())); |
| |
| |
| String dbquery = "CREATE DATABASE IF NOT EXISTS " + databaseName + " LOCATION '" + dbDir + "'"; |
| String tableQuery = "CREATE TABLE " + databaseName + "." + tableName + |
| "(key int, english string, spanish string) STORED BY " + |
| "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'" + |
| "TBLPROPERTIES ('" + HBaseConstants.PROPERTY_BULK_OUTPUT_MODE_KEY + "'='true'," + |
| "'hbase.columns.mapping'=':key," + familyName + ":english," + familyName + ":spanish')"; |
| |
| assertEquals(0, hcatDriver.run(dbquery).getResponseCode()); |
| assertEquals(0, hcatDriver.run(tableQuery).getResponseCode()); |
| |
| String data[] = {"1,english:ONE,spanish:UNO", |
| "2,english:TWO,spanish:DOS", |
| "3,english:THREE,spanish:TRES"}; |
| |
| // input/output settings |
| Path inputPath = new Path(methodTestDir, "mr_input"); |
| getFileSystem().mkdirs(inputPath); |
| //create multiple files so we can test with multiple mappers |
| for (int i = 0; i < data.length; i++) { |
| FSDataOutputStream os = getFileSystem().create(new Path(inputPath, "inputFile" + i + ".txt")); |
| os.write(Bytes.toBytes(data[i] + "\n")); |
| os.close(); |
| } |
| |
| //create job |
| Job job = new Job(conf, testName); |
| job.setWorkingDirectory(new Path(methodTestDir, "mr_work")); |
| job.setJarByClass(this.getClass()); |
| job.setMapperClass(MapHCatWrite.class); |
| |
| job.setInputFormatClass(TextInputFormat.class); |
| TextInputFormat.setInputPaths(job, inputPath); |
| |
| |
| job.setOutputFormatClass(HCatOutputFormat.class); |
| OutputJobInfo outputJobInfo = OutputJobInfo.create(databaseName, tableName, null); |
| HCatOutputFormat.setOutput(job, outputJobInfo); |
| |
| job.setMapOutputKeyClass(BytesWritable.class); |
| job.setMapOutputValueClass(HCatRecord.class); |
| |
| job.setOutputKeyClass(BytesWritable.class); |
| job.setOutputValueClass(HCatRecord.class); |
| |
| job.setNumReduceTasks(0); |
| |
| assertTrue(job.waitForCompletion(true)); |
| RevisionManager rm = HBaseRevisionManagerUtil.getOpenedRevisionManager(conf); |
| try { |
| TableSnapshot snapshot = rm.createSnapshot(databaseName + "." + tableName); |
| for (String el : snapshot.getColumnFamilies()) { |
| assertEquals(1, snapshot.getRevision(el)); |
| } |
| } finally { |
| rm.close(); |
| } |
| |
| //verify |
| HTable table = new HTable(conf, databaseName + "." + tableName); |
| Scan scan = new Scan(); |
| scan.addFamily(familyNameBytes); |
| ResultScanner scanner = table.getScanner(scan); |
| int index = 0; |
| for (Result result : scanner) { |
| String vals[] = data[index].toString().split(","); |
| for (int i = 1; i < vals.length; i++) { |
| String pair[] = vals[i].split(":"); |
| assertTrue(result.containsColumn(familyNameBytes, Bytes.toBytes(pair[0]))); |
| assertEquals(pair[1], Bytes.toString(result.getValue(familyNameBytes, Bytes.toBytes(pair[0])))); |
| assertEquals(1l, result.getColumn(familyNameBytes, Bytes.toBytes(pair[0])).get(0).getTimestamp()); |
| } |
| index++; |
| } |
| //test if load count is the same |
| assertEquals(data.length, index); |
| } |
| |
| @Test |
| public void bulkModeHCatOutputFormatTestWithDefaultDB() throws Exception { |
| String testName = "bulkModeHCatOutputFormatTestWithDefaultDB"; |
| Path methodTestDir = new Path(getTestDir(), testName); |
| |
| String databaseName = "default"; |
| String dbDir = new Path(methodTestDir, "DB_" + testName).toString(); |
| String tableName = newTableName(testName).toLowerCase(); |
| String familyName = "my_family"; |
| byte[] familyNameBytes = Bytes.toBytes(familyName); |
| |
| |
| //include hbase config in conf file |
| Configuration conf = new Configuration(allConf); |
| conf.set(HCatConstants.HCAT_KEY_HIVE_CONF, HCatUtil.serialize(allConf.getAllProperties())); |
| |
| |
| String dbquery = "CREATE DATABASE IF NOT EXISTS " + databaseName + " LOCATION '" + dbDir + "'"; |
| String tableQuery = "CREATE TABLE " + databaseName + "." + tableName + |
| "(key int, english string, spanish string) STORED BY " + |
| "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'" + |
| "TBLPROPERTIES ('" + HBaseConstants.PROPERTY_BULK_OUTPUT_MODE_KEY + "'='true'," + |
| "'hbase.columns.mapping'=':key," + familyName + ":english," + familyName + ":spanish')"; |
| |
| assertEquals(0, hcatDriver.run(dbquery).getResponseCode()); |
| assertEquals(0, hcatDriver.run(tableQuery).getResponseCode()); |
| |
| String data[] = {"1,english:ONE,spanish:UNO", |
| "2,english:TWO,spanish:DOS", |
| "3,english:THREE,spanish:TRES"}; |
| |
| // input/output settings |
| Path inputPath = new Path(methodTestDir, "mr_input"); |
| getFileSystem().mkdirs(inputPath); |
| FSDataOutputStream os = getFileSystem().create(new Path(inputPath, "inputFile.txt")); |
| for (String line : data) |
| os.write(Bytes.toBytes(line + "\n")); |
| os.close(); |
| |
| //create job |
| Job job = new Job(conf, testName); |
| job.setWorkingDirectory(new Path(methodTestDir, "mr_work")); |
| job.setJarByClass(this.getClass()); |
| job.setMapperClass(MapHCatWrite.class); |
| |
| job.setInputFormatClass(TextInputFormat.class); |
| TextInputFormat.setInputPaths(job, inputPath); |
| |
| |
| job.setOutputFormatClass(HCatOutputFormat.class); |
| OutputJobInfo outputJobInfo = OutputJobInfo.create(databaseName, tableName, null); |
| HCatOutputFormat.setOutput(job, outputJobInfo); |
| |
| job.setMapOutputKeyClass(BytesWritable.class); |
| job.setMapOutputValueClass(HCatRecord.class); |
| |
| job.setOutputKeyClass(BytesWritable.class); |
| job.setOutputValueClass(HCatRecord.class); |
| |
| job.setNumReduceTasks(0); |
| |
| assertTrue(job.waitForCompletion(true)); |
| |
| //verify |
| HTable table = new HTable(conf, tableName); |
| Scan scan = new Scan(); |
| scan.addFamily(familyNameBytes); |
| ResultScanner scanner = table.getScanner(scan); |
| int index = 0; |
| for (Result result : scanner) { |
| String vals[] = data[index].toString().split(","); |
| for (int i = 1; i < vals.length; i++) { |
| String pair[] = vals[i].split(":"); |
| assertTrue(result.containsColumn(familyNameBytes, Bytes.toBytes(pair[0]))); |
| assertEquals(pair[1], Bytes.toString(result.getValue(familyNameBytes, Bytes.toBytes(pair[0])))); |
| } |
| index++; |
| } |
| //test if load count is the same |
| assertEquals(data.length, index); |
| } |
| |
| @Test |
| public void bulkModeAbortTest() throws Exception { |
| String testName = "bulkModeAbortTest"; |
| Path methodTestDir = new Path(getTestDir(), testName); |
| String databaseName = testName.toLowerCase(); |
| String dbDir = new Path(methodTestDir, "DB_" + testName).toString(); |
| String tableName = newTableName(testName).toLowerCase(); |
| String familyName = "my_family"; |
| |
| // include hbase config in conf file |
| Configuration conf = new Configuration(allConf); |
| conf.set(HCatConstants.HCAT_KEY_HIVE_CONF, HCatUtil.serialize(allConf.getAllProperties())); |
| |
| String dbquery = "CREATE DATABASE IF NOT EXISTS " + databaseName + " LOCATION '" + dbDir |
| + "'"; |
| String tableQuery = "CREATE TABLE " + databaseName + "." + tableName + |
| "(key int, english string, spanish string) STORED BY " + |
| "'org.apache.hcatalog.hbase.HBaseHCatStorageHandler'" + |
| "TBLPROPERTIES ('" + HBaseConstants.PROPERTY_BULK_OUTPUT_MODE_KEY + "'='true'," + |
| "'hbase.columns.mapping'=':key," + familyName + ":english," + familyName |
| + ":spanish')"; |
| |
| assertEquals(0, hcatDriver.run(dbquery).getResponseCode()); |
| assertEquals(0, hcatDriver.run(tableQuery).getResponseCode()); |
| |
| String data[] = {"1,english:ONE,spanish:UNO", |
| "2,english:TWO,spanish:DOS", |
| "3,english:THREE,spanish:TRES"}; |
| |
| Path inputPath = new Path(methodTestDir, "mr_input"); |
| getFileSystem().mkdirs(inputPath); |
| // create multiple files so we can test with multiple mappers |
| for (int i = 0; i < data.length; i++) { |
| FSDataOutputStream os = getFileSystem().create( |
| new Path(inputPath, "inputFile" + i + ".txt")); |
| os.write(Bytes.toBytes(data[i] + "\n")); |
| os.close(); |
| } |
| |
| Path workingDir = new Path(methodTestDir, "mr_abort"); |
| OutputJobInfo outputJobInfo = OutputJobInfo.create(databaseName, |
| tableName, null); |
| Job job = configureJob(testName, |
| conf, workingDir, MapWriteAbortTransaction.class, |
| outputJobInfo, inputPath); |
| assertFalse(job.waitForCompletion(true)); |
| |
| // verify that revision manager has it as aborted transaction |
| RevisionManager rm = HBaseRevisionManagerUtil.getOpenedRevisionManager(conf); |
| try { |
| TableSnapshot snapshot = rm.createSnapshot(databaseName + "." + tableName); |
| for (String family : snapshot.getColumnFamilies()) { |
| assertEquals(1, snapshot.getRevision(family)); |
| List<FamilyRevision> abortedWriteTransactions = rm.getAbortedWriteTransactions( |
| databaseName + "." + tableName, family); |
| assertEquals(1, abortedWriteTransactions.size()); |
| assertEquals(1, abortedWriteTransactions.get(0).getRevision()); |
| } |
| } finally { |
| rm.close(); |
| } |
| |
| //verify that hbase does not have any of the records. |
| //Since records are only written during commitJob, |
| //hbase should not have any records. |
| HTable table = new HTable(conf, databaseName + "." + tableName); |
| Scan scan = new Scan(); |
| scan.addFamily(Bytes.toBytes(familyName)); |
| ResultScanner scanner = table.getScanner(scan); |
| assertFalse(scanner.iterator().hasNext()); |
| |
| // verify that the storage handler input format returns empty results. |
| Path outputDir = new Path(getTestDir(), |
| "mapred/testHBaseTableBulkIgnoreAbortedTransactions"); |
| FileSystem fs = getFileSystem(); |
| if (fs.exists(outputDir)) { |
| fs.delete(outputDir, true); |
| } |
| job = new Job(conf, "hbase-bulk-aborted-transaction"); |
| job.setJarByClass(this.getClass()); |
| job.setMapperClass(MapReadAbortedTransaction.class); |
| job.setInputFormatClass(HCatInputFormat.class); |
| InputJobInfo inputJobInfo = InputJobInfo.create(databaseName, |
| tableName, null); |
| HCatInputFormat.setInput(job, inputJobInfo); |
| job.setOutputFormatClass(TextOutputFormat.class); |
| TextOutputFormat.setOutputPath(job, outputDir); |
| job.setMapOutputKeyClass(BytesWritable.class); |
| job.setMapOutputValueClass(Text.class); |
| job.setOutputKeyClass(BytesWritable.class); |
| job.setOutputValueClass(Text.class); |
| job.setNumReduceTasks(0); |
| assertTrue(job.waitForCompletion(true)); |
| } |
| |
| private Job configureJob(String jobName, Configuration conf, |
| Path workingDir, Class<? extends Mapper> mapperClass, |
| OutputJobInfo outputJobInfo, Path inputPath) throws IOException { |
| Job job = new Job(conf, jobName); |
| job.setWorkingDirectory(workingDir); |
| job.setJarByClass(this.getClass()); |
| job.setMapperClass(mapperClass); |
| |
| job.setInputFormatClass(TextInputFormat.class); |
| TextInputFormat.setInputPaths(job, inputPath); |
| job.setOutputFormatClass(HCatOutputFormat.class); |
| HCatOutputFormat.setOutput(job, outputJobInfo); |
| |
| job.setMapOutputKeyClass(BytesWritable.class); |
| job.setMapOutputValueClass(HCatRecord.class); |
| job.setOutputKeyClass(BytesWritable.class); |
| job.setOutputValueClass(HCatRecord.class); |
| |
| job.setNumReduceTasks(0); |
| return job; |
| } |
| |
| } |
| |
