wchevreuil commented on code in PR #163:
URL: https://github.com/apache/hbase-connectors/pull/163#discussion_r3967396019
##########
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:
I don't think this is needed.
--
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]
