Sigma-Ma commented on code in PR #163: URL: https://github.com/apache/hbase-connectors/pull/163#discussion_r3985485034
########## spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/BulkLoadSuite.scala: ########## @@ -0,0 +1,1061 @@ +/* + * 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.hadoop.hbase.spark + +import java.io.File +import java.net.URI +import java.nio.file.Files +import org.apache.hadoop.fs.{FileSystem, Path} +import org.apache.hadoop.hbase.{CellUtil, HBaseTestingUtility, HConstants, TableName} +import org.apache.hadoop.hbase.client.{ConnectionFactory, Get} +import org.apache.hadoop.hbase.io.hfile.{CacheConfig, HFile} +import org.apache.hadoop.hbase.spark.HBaseRDDFunctions._ +import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles +import org.apache.hadoop.hbase.util.Bytes +import org.apache.spark.{SparkConf, SparkContext} +import org.junit.rules.TemporaryFolder +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} +import org.scalatest.funsuite.AnyFunSuite + +class BulkLoadSuite + extends AnyFunSuite + with BeforeAndAfterEach + with BeforeAndAfterAll + with Logging { + @transient var sc: SparkContext = null + var TEST_UTIL = new HBaseTestingUtility + + val tableName = "t1" + val columnFamily1 = "f1" + val columnFamily2 = "f2" + val testFolder = new TemporaryFolder() + + override def beforeAll(): Unit = { + TEST_UTIL.startMiniCluster() + logInfo(" - minicluster started") + + try { + TEST_UTIL.deleteTable(TableName.valueOf(tableName)) + } catch { + case e: Exception => + logInfo(" - no table " + tableName + " found") + } + + logInfo(" - created table") + + val sparkConf = new SparkConf() + .setMaster("local[2]") + .setAppName("BulkLoadSuite") + .set("spark.hadoopRDD.ignoreEmptySplits", "false") + sc = new SparkContext(sparkConf) + } + + override def afterAll(): Unit = { + logInfo("shuting down minicluster") + TEST_UTIL.shutdownMiniCluster() + logInfo(" - minicluster shut down") + TEST_UTIL.cleanupTestDir() + sc.stop() + } + + test("Staging dir: Test usage of staging dir on a separate filesystem") { + val config = TEST_UTIL.getConfiguration + + logInfo(" - creating table " + tableName) + TEST_UTIL.createTable( + TableName.valueOf(tableName), + Array(Bytes.toBytes(columnFamily1), Bytes.toBytes(columnFamily2))) + + // Test creates rdd with 2 column families and + // write those to hfiles on local filesystem + // using bulkLoad functionality. We don't check the load functionality + // due the limitations of the HBase Minicluster + + val rdd = sc.parallelize( + Array[(Array[Byte], (Array[Byte], Array[Byte], Array[Byte]))]( + ( + Bytes.toBytes("1"), + (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))), + ( + Bytes.toBytes("2"), + (Bytes.toBytes(columnFamily2), Bytes.toBytes("b"), Bytes.toBytes("bar.2"))))) + + val hbaseContext = new HBaseContext(sc, config) + val uri = Files.createTempDirectory("tmpDirPrefix").toUri + val stagingUri = new URI(uri + "staging_dir") + val stagingFolder = new File(stagingUri) + val fs = new Path(stagingUri.toString).getFileSystem(config) + try { + hbaseContext.bulkLoad[(Array[Byte], (Array[Byte], Array[Byte], Array[Byte]))]( + rdd, + TableName.valueOf(tableName), + t => { + val rowKey = t._1 + val family: Array[Byte] = t._2._1 + val qualifier = t._2._2 + val value: Array[Byte] = t._2._3 + + val keyFamilyQualifier = new KeyFamilyQualifier(rowKey, family, qualifier) + + Seq((keyFamilyQualifier, value)).iterator + }, + stagingUri.toString) + + assert(fs.listStatus(new Path(stagingFolder.getPath)).length == 2) + + } finally { + val admin = ConnectionFactory.createConnection(config).getAdmin + try { + admin.disableTable(TableName.valueOf(tableName)) + admin.deleteTable(TableName.valueOf(tableName)) + } finally { + admin.close() + } + fs.delete(new Path(stagingFolder.getPath), true) + + testFolder.delete() + + } + } + + test( + "Wide Row Bulk Load: Test multi family and multi column tests " + + "with all default HFile Configs.") { + val config = TEST_UTIL.getConfiguration + + logInfo(" - creating table " + tableName) + TEST_UTIL.createTable( + TableName.valueOf(tableName), + Array(Bytes.toBytes(columnFamily1), Bytes.toBytes(columnFamily2))) + + // There are a number of tests in here. + // 1. Row keys are not in order + // 2. Qualifiers are not in order + // 3. Column Families are not in order + // 4. There are tests for records in one column family and some in two column families + // 5. There are records will a single qualifier and some with two + val rdd = sc.parallelize( + Array[(Array[Byte], (Array[Byte], Array[Byte], Array[Byte]))]( + ( + Bytes.toBytes("1"), + (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))), + ( + Bytes.toBytes("3"), + (Bytes.toBytes(columnFamily2), Bytes.toBytes("b"), Bytes.toBytes("foo2.a"))), + ( + Bytes.toBytes("3"), + (Bytes.toBytes(columnFamily2), Bytes.toBytes("a"), Bytes.toBytes("foo2.b"))), + ( + Bytes.toBytes("3"), + (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo2.c"))), + ( + Bytes.toBytes("5"), + (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo3"))), + ( + Bytes.toBytes("4"), + (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo.1"))), + ( + Bytes.toBytes("4"), + (Bytes.toBytes(columnFamily2), Bytes.toBytes("b"), Bytes.toBytes("foo.2"))), + ( + Bytes.toBytes("2"), + (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("bar.1"))), + ( + Bytes.toBytes("2"), + (Bytes.toBytes(columnFamily1), Bytes.toBytes("b"), Bytes.toBytes("bar.2"))))) + + val hbaseContext = new HBaseContext(sc, config) + + testFolder.create() + val stagingFolder = testFolder.newFolder() + + hbaseContext.bulkLoad[(Array[Byte], (Array[Byte], Array[Byte], Array[Byte]))]( + rdd, + TableName.valueOf(tableName), + t => { + val rowKey = t._1 + val family: Array[Byte] = t._2._1 + val qualifier = t._2._2 + val value: Array[Byte] = t._2._3 + + val keyFamilyQualifier = new KeyFamilyQualifier(rowKey, family, qualifier) + + Seq((keyFamilyQualifier, value)).iterator + }, + stagingFolder.getPath) + + val fs = FileSystem.get(config) + assert(fs.listStatus(new Path(stagingFolder.getPath)).length == 2) + + val conn = ConnectionFactory.createConnection(config) + + val load = new LoadIncrementalHFiles(config) + val table = conn.getTable(TableName.valueOf(tableName)) + try { + load.doBulkLoad( + new Path(stagingFolder.getPath), + conn.getAdmin, + table, + conn.getRegionLocator(TableName.valueOf(tableName))) + + val cells5 = table.get(new Get(Bytes.toBytes("5"))).listCells() + assert(cells5.size == 1) + assert(Bytes.toString(CellUtil.cloneValue(cells5.get(0))).equals("foo3")) + assert(Bytes.toString(CellUtil.cloneFamily(cells5.get(0))).equals("f1")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells5.get(0))).equals("a")) + + val cells4 = table.get(new Get(Bytes.toBytes("4"))).listCells() + assert(cells4.size == 2) + assert(Bytes.toString(CellUtil.cloneValue(cells4.get(0))).equals("foo.1")) + assert(Bytes.toString(CellUtil.cloneFamily(cells4.get(0))).equals("f1")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells4.get(0))).equals("a")) + assert(Bytes.toString(CellUtil.cloneValue(cells4.get(1))).equals("foo.2")) + assert(Bytes.toString(CellUtil.cloneFamily(cells4.get(1))).equals("f2")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells4.get(1))).equals("b")) + + val cells3 = table.get(new Get(Bytes.toBytes("3"))).listCells() + assert(cells3.size == 3) + assert(Bytes.toString(CellUtil.cloneValue(cells3.get(0))).equals("foo2.c")) + assert(Bytes.toString(CellUtil.cloneFamily(cells3.get(0))).equals("f1")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells3.get(0))).equals("a")) + assert(Bytes.toString(CellUtil.cloneValue(cells3.get(1))).equals("foo2.b")) + assert(Bytes.toString(CellUtil.cloneFamily(cells3.get(1))).equals("f2")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells3.get(1))).equals("a")) + assert(Bytes.toString(CellUtil.cloneValue(cells3.get(2))).equals("foo2.a")) + assert(Bytes.toString(CellUtil.cloneFamily(cells3.get(2))).equals("f2")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells3.get(2))).equals("b")) + + val cells2 = table.get(new Get(Bytes.toBytes("2"))).listCells() + assert(cells2.size == 2) + assert(Bytes.toString(CellUtil.cloneValue(cells2.get(0))).equals("bar.1")) + assert(Bytes.toString(CellUtil.cloneFamily(cells2.get(0))).equals("f1")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells2.get(0))).equals("a")) + assert(Bytes.toString(CellUtil.cloneValue(cells2.get(1))).equals("bar.2")) + assert(Bytes.toString(CellUtil.cloneFamily(cells2.get(1))).equals("f1")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells2.get(1))).equals("b")) + + val cells1 = table.get(new Get(Bytes.toBytes("1"))).listCells() + assert(cells1.size == 1) + assert(Bytes.toString(CellUtil.cloneValue(cells1.get(0))).equals("foo1")) + assert(Bytes.toString(CellUtil.cloneFamily(cells1.get(0))).equals("f1")) + assert(Bytes.toString(CellUtil.cloneQualifier(cells1.get(0))).equals("a")) + + } finally { + table.close() + val admin = ConnectionFactory.createConnection(config).getAdmin + try { + admin.disableTable(TableName.valueOf(tableName)) + admin.deleteTable(TableName.valueOf(tableName)) + } finally { + admin.close() + } Review Comment: Fixed. Each test now uses one connection and closes the Admin, Table, RegionLocator, and Connection during cleanup. I also fixed the same pattern in the other bulk-load tests. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
