wchevreuil commented on code in PR #163:
URL: https://github.com/apache/hbase-connectors/pull/163#discussion_r3967416881
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:
##########
@@ -137,10 +151,13 @@ class HBaseContext(@transient val sc: SparkContext,
@transient val config: Confi
* HBase Deletes
* @param batchSize The number of delete to batch before sending to
HBase
*/
- def bulkDelete[T](rdd: RDD[T], tableName: TableName, f: (T) => Delete,
batchSize: Integer): Unit =
- {
- bulkMutation(rdd, tableName, f, batchSize)
- }
+ def bulkDelete[T](
+ rdd: RDD[T],
+ tableName: TableName,
+ f: (T) => Delete,
+ batchSize: Integer): Unit = {
+ bulkMutation(rdd, tableName, f, batchSize)
+ }
Review Comment:
can we revert these non-logic changes?
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:
##########
@@ -310,12 +327,13 @@ class HBaseContext(@transient val sc: SparkContext,
@transient val config: Confi
val table = connection.getTable(TableName.valueOf(tName))
try {
val mutationList = new java.util.ArrayList[Mutation]()
- iterator.foreach { t =>
- mutationList.add(f(t))
- if (mutationList.size >= batchSize) {
- table.batch(mutationList, null)
- mutationList.clear()
- }
+ iterator.foreach {
+ t =>
+ mutationList.add(f(t))
+ if (mutationList.size >= batchSize) {
+ table.batch(mutationList, null)
+ mutationList.clear()
+ }
Review Comment:
can we revert these non-logic changes?
##########
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:
Yes, despite this being used on the spark3 module, let's take the
opportunity to address this in the spark4 module.
--
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]
