Copilot commented on code in PR #163: URL: https://github.com/apache/hbase-connectors/pull/163#discussion_r3966769405
########## 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))) Review Comment: The test creates an HBase `Connection` (`conn`) but never closes it, and also creates a *second* connection just to get an `Admin` in the `finally` block (closing only the `Admin` does not close the underlying `Connection`). This can leak threads/resources and cause test flakiness/hangs. Use a single `Connection` per test and ensure it’s closed in `finally`; derive `Admin`/`RegionLocator`/`Table` from that connection and close them appropriately. ########## 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: The test creates an HBase `Connection` (`conn`) but never closes it, and also creates a *second* connection just to get an `Admin` in the `finally` block (closing only the `Admin` does not close the underlying `Connection`). This can leak threads/resources and cause test flakiness/hangs. Use a single `Connection` per test and ensure it’s closed in `finally`; derive `Admin`/`RegionLocator`/`Table` from that connection and close them appropriately. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/BulkLoadPartitioner.scala: ########## @@ -0,0 +1,61 @@ +/* + * 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.util +import java.util.Comparator +import org.apache.hadoop.hbase.util.Bytes +import org.apache.spark.Partitioner +import org.apache.yetus.audience.InterfaceAudience + +/** + * A Partitioner implementation that will separate records to different + * HBase Regions based on region splits + * + * @param startKeys The start keys for the given table + */ [email protected] +class BulkLoadPartitioner(startKeys: Array[Array[Byte]]) extends Partitioner { + // when table not exist, startKeys = Byte[0][] + override def numPartitions: Int = if (startKeys.length == 0) 1 else startKeys.length + + override def getPartition(key: Any): Int = { + + val comparator: Comparator[Array[Byte]] = new Comparator[Array[Byte]] { + override def compare(o1: Array[Byte], o2: Array[Byte]): Int = { + Bytes.compareTo(o1, o2) + } + } Review Comment: Creating a new `Comparator` inside `getPartition` allocates per record and can become a hotspot during the bulk-load shuffle. Make the comparator a `private val` on the class (or use an existing comparator) so it’s reused across calls. Also consider overriding `equals`/`hashCode` for this `Partitioner` (based on `startKeys`) so Spark can recognize identical partitioners and avoid unnecessary shuffles. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/FamilyHFileWriteOptions.scala: ########## @@ -0,0 +1,39 @@ +/* + * 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.Serializable +import org.apache.yetus.audience.InterfaceAudience; Review Comment: Trailing semicolons in Scala imports are non-idiomatic and may conflict with formatting/lint expectations (especially since the PR mentions Spotless formatting). Remove the trailing `;` to match typical Scala style. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/BulkLoadPartitioner.scala: ########## @@ -0,0 +1,61 @@ +/* + * 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.util +import java.util.Comparator +import org.apache.hadoop.hbase.util.Bytes +import org.apache.spark.Partitioner +import org.apache.yetus.audience.InterfaceAudience + +/** + * A Partitioner implementation that will separate records to different + * HBase Regions based on region splits + * + * @param startKeys The start keys for the given table + */ [email protected] +class BulkLoadPartitioner(startKeys: Array[Array[Byte]]) extends Partitioner { + // when table not exist, startKeys = Byte[0][] + override def numPartitions: Int = if (startKeys.length == 0) 1 else startKeys.length + + override def getPartition(key: Any): Int = { + + val comparator: Comparator[Array[Byte]] = new Comparator[Array[Byte]] { + override def compare(o1: Array[Byte], o2: Array[Byte]): Int = { + Bytes.compareTo(o1, o2) + } + } + + val rowKey: Array[Byte] = + key match { + case qualifier: KeyFamilyQualifier => + qualifier.rowKey + case wrapper: ByteArrayWrapper => + wrapper.value + case _ => + key.asInstanceOf[Array[Byte]] + } + var partition = util.Arrays.binarySearch(startKeys, rowKey, comparator) Review Comment: Creating a new `Comparator` inside `getPartition` allocates per record and can become a hotspot during the bulk-load shuffle. Make the comparator a `private val` on the class (or use an existing comparator) so it’s reused across calls. Also consider overriding `equals`/`hashCode` for this `Partitioner` (based on `startKeys`) so Spark can recognize identical partitioners and avoid unnecessary shuffles. ########## 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") Review Comment: Typo in log message: `shuting` should be `shutting`. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala: ########## @@ -364,6 +382,560 @@ class HBaseContext(@transient val sc: SparkContext, @transient val config: Confi } } + /** + * Spark Implementation of HBase Bulk load for wide rows or when + * values are not already combined at the time of the map process + * + * This will take the content from an existing RDD then sort and shuffle + * it with respect to region splits. The result of that sort and shuffle + * will be written to HFiles. + * + * After this function is executed the user will have to call + * LoadIncrementalHFiles.doBulkLoad(...) to move the files into HBase + * + * Also note this version of bulk load is different from past versions in + * that it includes the qualifier as part of the sort process. The + * reason for this is to be able to support rows will very large number + * of columns. + * + * @param rdd The RDD we are bulk loading from + * @param tableName The HBase table we are loading into + * @param flatMap A flapMap function that will make every Review Comment: Correct typo in Scaladoc: `flapMap` should be `flatMap`. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseRDDFunctions.scala: ########## @@ -169,5 +170,102 @@ object HBaseRDDFunctions { f: (Iterator[T], Connection) => Iterator[R]): RDD[R] = { hc.mapPartitions[T, R](rdd, f) } + + /** + * Spark Implementation of HBase Bulk load for wide rows or when + * values are not already combined at the time of the map process + * + * A Spark Implementation of HBase Bulk load + * + * This will take the content from an existing RDD then sort and shuffle + * it with respect to region splits. The result of that sort and shuffle + * will be written to HFiles. + * + * After this function is executed the user will have to call + * LoadIncrementalHFiles.doBulkLoad(...) to move the files into HBase + * + * Also note this version of bulk load is different from past versions in + * that it includes the qualifier as part of the sort process. The + * reason for this is to be able to support rows will very large number + * of columns. + * + * @param tableName The HBase table we are loading into + * @param flatMap A flapMap function that will make every row in the RDD Review Comment: Correct typo in Scaladoc: `flapMap` should be `flatMap`. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala: ########## @@ -364,6 +382,560 @@ class HBaseContext(@transient val sc: SparkContext, @transient val config: Confi } } + /** + * Spark Implementation of HBase Bulk load for wide rows or when + * values are not already combined at the time of the map process + * + * This will take the content from an existing RDD then sort and shuffle + * it with respect to region splits. The result of that sort and shuffle + * will be written to HFiles. + * + * After this function is executed the user will have to call + * LoadIncrementalHFiles.doBulkLoad(...) to move the files into HBase + * + * Also note this version of bulk load is different from past versions in + * that it includes the qualifier as part of the sort process. The + * reason for this is to be able to support rows will very large number + * of columns. + * + * @param rdd The RDD we are bulk loading from + * @param tableName The HBase table we are loading into + * @param flatMap A flapMap function that will make every + * row in the RDD + * into N cells for the bulk load + * @param stagingDir The location on the FileSystem to bulk load into + * @param familyHFileWriteOptionsMap Options that will define how the HFile for a + * column family is written + * @param compactionExclude Compaction excluded for the HFiles + * @param maxSize Max size for the HFiles before they roll + * @param nowTimeStamp Version timestamp + * @tparam T The Type of values in the original RDD + */ + def bulkLoad[T]( + rdd: RDD[T], + tableName: TableName, + flatMap: (T) => Iterator[(KeyFamilyQualifier, Array[Byte])], + stagingDir: String, + familyHFileWriteOptionsMap: util.Map[Array[Byte], FamilyHFileWriteOptions] = + new util.HashMap[Array[Byte], FamilyHFileWriteOptions], + compactionExclude: Boolean = false, + maxSize: Long = HConstants.DEFAULT_MAX_FILE_SIZE, + nowTimeStamp: Long = System.currentTimeMillis()): Unit = { + val stagingPath = new Path(stagingDir) + val fs = stagingPath.getFileSystem(config) + if (fs.exists(stagingPath)) { + throw new FileAlreadyExistsException("Path " + stagingDir + " already exists") + } Review Comment: New behavior is introduced here (failing fast if the staging path already exists), but `BulkLoadSuite` doesn’t appear to assert this error path. Please add a test that pre-creates `stagingDir` and verifies `bulkLoad`/`bulkLoadThinRows` throws `FileAlreadyExistsException` with a helpful message. ########## spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala: ########## @@ -364,6 +382,560 @@ class HBaseContext(@transient val sc: SparkContext, @transient val config: Confi } } + /** + * Spark Implementation of HBase Bulk load for wide rows or when + * values are not already combined at the time of the map process + * + * This will take the content from an existing RDD then sort and shuffle + * it with respect to region splits. The result of that sort and shuffle + * will be written to HFiles. + * + * After this function is executed the user will have to call + * LoadIncrementalHFiles.doBulkLoad(...) to move the files into HBase + * + * Also note this version of bulk load is different from past versions in + * that it includes the qualifier as part of the sort process. The + * reason for this is to be able to support rows will very large number + * of columns. + * + * @param rdd The RDD we are bulk loading from + * @param tableName The HBase table we are loading into + * @param flatMap A flapMap function that will make every + * row in the RDD + * into N cells for the bulk load + * @param stagingDir The location on the FileSystem to bulk load into + * @param familyHFileWriteOptionsMap Options that will define how the HFile for a + * column family is written + * @param compactionExclude Compaction excluded for the HFiles + * @param maxSize Max size for the HFiles before they roll + * @param nowTimeStamp Version timestamp + * @tparam T The Type of values in the original RDD + */ + def bulkLoad[T]( + rdd: RDD[T], + tableName: TableName, + flatMap: (T) => Iterator[(KeyFamilyQualifier, Array[Byte])], + stagingDir: String, + familyHFileWriteOptionsMap: util.Map[Array[Byte], FamilyHFileWriteOptions] = + new util.HashMap[Array[Byte], FamilyHFileWriteOptions], + compactionExclude: Boolean = false, + maxSize: Long = HConstants.DEFAULT_MAX_FILE_SIZE, + nowTimeStamp: Long = System.currentTimeMillis()): Unit = { + val stagingPath = new Path(stagingDir) + val fs = stagingPath.getFileSystem(config) + if (fs.exists(stagingPath)) { + throw new FileAlreadyExistsException("Path " + stagingDir + " already exists") + } + val conn = HBaseConnectionCache.getConnection(config) + try { + val regionLocator = conn.getRegionLocator(tableName) + val startKeys = regionLocator.getStartKeys + if (startKeys.length == 0) { + logInfo("Table " + tableName.toString + " was not found") + } + val defaultCompressionStr = + config.get("hfile.compression", Compression.Algorithm.NONE.getName) + val hfileCompression = HFileWriterImpl + .compressionByName(defaultCompressionStr) + val tableRawName = tableName.getName + + val familyHFileWriteOptionsMapInternal = + new util.HashMap[ByteArrayWrapper, FamilyHFileWriteOptions] + + val entrySetIt = familyHFileWriteOptionsMap.entrySet().iterator() + + while (entrySetIt.hasNext) { + val entry = entrySetIt.next() + familyHFileWriteOptionsMapInternal.put(new ByteArrayWrapper(entry.getKey), entry.getValue) + } + + val regionSplitPartitioner = + new BulkLoadPartitioner(startKeys) + + // This is where all the magic happens + // Here we are going to do the following things + // 1. FlapMap every row in the RDD into key column value tuples + // 2. Then we are going to repartition sort and shuffle + // 3. Finally we are going to write out our HFiles + rdd + .flatMap(r => flatMap(r)) + .repartitionAndSortWithinPartitions(regionSplitPartitioner) + .hbaseForeachPartition( + this, + (it, conn) => { + + val conf = broadcastedConf.value.value + val fs = new Path(stagingDir).getFileSystem(conf) + val writerMap = new mutable.HashMap[ByteArrayWrapper, WriterLength] + var previousRow: Array[Byte] = HConstants.EMPTY_BYTE_ARRAY + var rollOverRequested = false + val localTableName = TableName.valueOf(tableRawName) + + // Here is where we finally iterate through the data in this partition of the + // RDD that has been sorted and partitioned + it.foreach { + case (keyFamilyQualifier, cellValue: Array[Byte]) => + val wl = writeValueToHFile( + keyFamilyQualifier.rowKey, + keyFamilyQualifier.family, + keyFamilyQualifier.qualifier, + cellValue, + nowTimeStamp, + fs, + conn, + localTableName, + conf, + familyHFileWriteOptionsMapInternal, + hfileCompression, + writerMap, + stagingDir) + + rollOverRequested = rollOverRequested || wl.written > maxSize + + // This will only roll if we have at least one column family file that is + // bigger then maxSize and we have finished a given row key + if (rollOverRequested && Bytes + .compareTo(previousRow, keyFamilyQualifier.rowKey) != 0) { + rollWriters(fs, writerMap, regionSplitPartitioner, previousRow, compactionExclude) + rollOverRequested = false + } + + previousRow = keyFamilyQualifier.rowKey + } + // We have finished all the data so lets close up the writers + rollWriters(fs, writerMap, regionSplitPartitioner, previousRow, compactionExclude) + rollOverRequested = false + }) + } finally { + if (null != conn) conn.close() + } + } + + /** + * Spark Implementation of HBase Bulk load for short rows some where less then + * a 1000 columns. This bulk load should be faster for tables will thinner + * rows then the other spark implementation of bulk load that puts only one + * value into a record going into a shuffle + * + * This will take the content from an existing RDD then sort and shuffle + * it with respect to region splits. The result of that sort and shuffle + * will be written to HFiles. + * + * After this function is executed the user will have to call + * LoadIncrementalHFiles.doBulkLoad(...) to move the files into HBase + * + * In this implementation, only the rowKey is given to the shuffle as the key + * and all the columns are already linked to the RowKey before the shuffle + * stage. The sorting of the qualifier is done in memory out side of the + * shuffle stage + * + * Also make sure that incoming RDDs only have one record for every row key. + * + * @param rdd The RDD we are bulk loading from + * @param tableName The HBase table we are loading into + * @param mapFunction A function that will convert the RDD records to + * the key value format used for the shuffle to prep + * for writing to the bulk loaded HFiles + * @param stagingDir The location on the FileSystem to bulk load into + * @param familyHFileWriteOptionsMap Options that will define how the HFile for a + * column family is written + * @param compactionExclude Compaction excluded for the HFiles + * @param maxSize Max size for the HFiles before they roll + * @tparam T The Type of values in the original RDD + */ + def bulkLoadThinRows[T]( + rdd: RDD[T], + tableName: TableName, + mapFunction: (T) => (ByteArrayWrapper, FamiliesQualifiersValues), + stagingDir: String, + familyHFileWriteOptionsMap: util.Map[Array[Byte], FamilyHFileWriteOptions] = + new util.HashMap[Array[Byte], FamilyHFileWriteOptions], + compactionExclude: Boolean = false, + maxSize: Long = HConstants.DEFAULT_MAX_FILE_SIZE): Unit = { + val stagingPath = new Path(stagingDir) + val fs = stagingPath.getFileSystem(config) + if (fs.exists(stagingPath)) { + throw new FileAlreadyExistsException("Path " + stagingDir + " already exists") + } + val conn = HBaseConnectionCache.getConnection(config) + try { + val regionLocator = conn.getRegionLocator(tableName) + val startKeys = regionLocator.getStartKeys + if (startKeys.length == 0) { + logInfo("Table " + tableName.toString + " was not found") + } + val defaultCompressionStr = + config.get("hfile.compression", Compression.Algorithm.NONE.getName) + val defaultCompression = HFileWriterImpl + .compressionByName(defaultCompressionStr) + val nowTimeStamp = System.currentTimeMillis() + val tableRawName = tableName.getName + + val familyHFileWriteOptionsMapInternal = + new util.HashMap[ByteArrayWrapper, FamilyHFileWriteOptions] + + val entrySetIt = familyHFileWriteOptionsMap.entrySet().iterator() + + while (entrySetIt.hasNext) { + val entry = entrySetIt.next() + familyHFileWriteOptionsMapInternal.put(new ByteArrayWrapper(entry.getKey), entry.getValue) + } + + val regionSplitPartitioner = + new BulkLoadPartitioner(startKeys) + + // This is where all the magic happens + // Here we are going to do the following things + // 1. FlapMap every row in the RDD into key column value tuples + // 2. Then we are going to repartition sort and shuffle + // 3. Finally we are going to write out our HFiles + rdd + .map(r => mapFunction(r)) + .repartitionAndSortWithinPartitions(regionSplitPartitioner) + .hbaseForeachPartition( + this, + (it, conn) => { + + val conf = broadcastedConf.value.value + val fs = new Path(stagingDir).getFileSystem(conf) + val writerMap = new mutable.HashMap[ByteArrayWrapper, WriterLength] + var previousRow: Array[Byte] = HConstants.EMPTY_BYTE_ARRAY + var rollOverRequested = false + val localTableName = TableName.valueOf(tableRawName) + + // Here is where we finally iterate through the data in this partition of the + // RDD that has been sorted and partitioned + it.foreach { + case (rowKey: ByteArrayWrapper, familiesQualifiersValues: FamiliesQualifiersValues) => + if (Bytes.compareTo(previousRow, rowKey.value) == 0) { + throw new KeyAlreadyExistsException( + "The following key was sent to the " + + "HFile load more then one: " + Bytes.toString(previousRow)) + } Review Comment: `javax.management.openmbean.KeyAlreadyExistsException` is a JMX/OpenMBean-specific exception and is a confusing choice for data-validation in a bulk-load API. Prefer a standard exception type (e.g., `IllegalArgumentException`/`IllegalStateException`) or a connector-specific exception, and update the message grammar (\"more than once\"). -- 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]
