You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by jm...@apache.org on 2016/01/26 20:27:19 UTC
hbase git commit: HBASE-15036 Update HBase Spark documentation to
include bulk load with thin records (Ted Malaska)
Repository: hbase
Updated Branches:
refs/heads/master 4ca27a68e -> cb539917a
HBASE-15036 Update HBase Spark documentation to include bulk load with thin records (Ted Malaska)
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/cb539917
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/cb539917
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/cb539917
Branch: refs/heads/master
Commit: cb539917a2cb307060b0ea4330a52cd40fe7e150
Parents: 4ca27a6
Author: Jonathan M Hsieh <jm...@apache.org>
Authored: Tue Jan 26 11:02:48 2016 -0800
Committer: Jonathan M Hsieh <jm...@apache.org>
Committed: Tue Jan 26 11:02:48 2016 -0800
----------------------------------------------------------------------
src/main/asciidoc/_chapters/spark.adoc | 89 +++++++++++++++++++++++++----
1 file changed, 78 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/cb539917/src/main/asciidoc/_chapters/spark.adoc
----------------------------------------------------------------------
diff --git a/src/main/asciidoc/_chapters/spark.adoc b/src/main/asciidoc/_chapters/spark.adoc
index 37503e9..b1bdb5d 100644
--- a/src/main/asciidoc/_chapters/spark.adoc
+++ b/src/main/asciidoc/_chapters/spark.adoc
@@ -210,19 +210,25 @@ to the HBase Connections in the executors
== Bulk Load
-Spark bulk load follows very closely to the MapReduce implementation of bulk
-load. In short, a partitioner partitions based on region splits and
+There are two options for bulk loading data into HBase with Spark. There is the
+basic bulk load functionality that will work for cases where your rows have
+millions of columns and cases where your columns are not consolidated and
+partitions before the on the map side of the Spark bulk load process.
+
+There is also a thin record bulk load option with Spark, this second option is
+designed for tables that have less then 10k columns per row. The advantage
+of this second option is higher throughput and less over all load on the Spark
+shuffle operation.
+
+Both implementations work more or less like the MapReduce bulk load process in
+that a partitioner partitions the rowkeys based on region splits and
the row keys are sent to the reducers in order, so that HFiles can be written
-out. In Spark terms, the bulk load will be focused around a
-`repartitionAndSortWithinPartitions` followed by a `foreachPartition`.
+out directly from the reduce phase.
-The only major difference with the Spark implementation compared to the
-MapReduce implementation is that the column qualifier is included in the shuffle
-ordering process. This was done because the MapReduce bulk load implementation
-would have memory issues with loading rows with a large numbers of columns, as a
-result of the sorting of those columns being done in the memory of the reducer JVM.
-Instead, that ordering is done in the Spark Shuffle, so there should no longer
-be a limit to the number of columns in a row for bulk loading.
+In Spark terms, the bulk load will be implemented around a the Spark
+`repartitionAndSortWithinPartitions` followed by a Spark `foreachPartition`.
+
+First lets look at an example of using the basic bulk load functionality
.Bulk Loading Example
====
@@ -237,6 +243,11 @@ val config = new HBaseConfiguration()
val hbaseContext = new HBaseContext(sc, config)
val stagingFolder = ...
+val rdd = sc.parallelize(Array(
+ (Bytes.toBytes("1"),
+ (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))),
+ (Bytes.toBytes("3"),
+ (Bytes.toBytes(columnFamily1), Bytes.toBytes("b"), Bytes.toBytes("foo2.b"))), ...
rdd.hbaseBulkLoad(TableName.valueOf(tableName),
t => {
@@ -290,6 +301,11 @@ val config = new HBaseConfiguration()
val hbaseContext = new HBaseContext(sc, config)
val stagingFolder = ...
+val rdd = sc.parallelize(Array(
+ (Bytes.toBytes("1"),
+ (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))),
+ (Bytes.toBytes("3"),
+ (Bytes.toBytes(columnFamily1), Bytes.toBytes("b"), Bytes.toBytes("foo2.b"))), ...
val familyHBaseWriterOptions = new java.util.HashMap[Array[Byte], FamilyHFileWriteOptions]
val f1Options = new FamilyHFileWriteOptions("GZ", "ROW", 128, "PREFIX")
@@ -318,6 +334,57 @@ load.doBulkLoad(new Path(stagingFolder.getPath),
----
====
+Now lets look at how you would call the thin record bulk load implementation
+
+.Using thin record bulk load
+====
+
+[source, scala]
+----
+val sc = new SparkContext("local", "test")
+val config = new HBaseConfiguration()
+
+val hbaseContext = new HBaseContext(sc, config)
+
+val stagingFolder = ...
+val rdd = sc.parallelize(Array(
+ ("1",
+ (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))),
+ ("3",
+ (Bytes.toBytes(columnFamily1), Bytes.toBytes("b"), Bytes.toBytes("foo2.b"))), ...
+
+rdd.hbaseBulkLoadThinRows(hbaseContext,
+ TableName.valueOf(tableName),
+ t => {
+ val rowKey = t._1
+
+ val familyQualifiersValues = new FamiliesQualifiersValues
+ t._2.foreach(f => {
+ val family:Array[Byte] = f._1
+ val qualifier = f._2
+ val value:Array[Byte] = f._3
+
+ familyQualifiersValues +=(family, qualifier, value)
+ })
+ (new ByteArrayWrapper(Bytes.toBytes(rowKey)), familyQualifiersValues)
+ },
+ stagingFolder.getPath,
+ new java.util.HashMap[Array[Byte], FamilyHFileWriteOptions],
+ compactionExclude = false,
+ 20)
+
+val load = new LoadIncrementalHFiles(config)
+load.doBulkLoad(new Path(stagingFolder.getPath),
+ conn.getAdmin, table, conn.getRegionLocator(TableName.valueOf(tableName)))
+----
+====
+
+Note that the big difference in using bulk load for thin rows is the function
+returns a tuple with the first value being the row key and the second value
+being an object of FamiliesQualifiersValues, which will contain all the
+values for this row for all column families.
+
+
== SparkSQL/DataFrames
http://spark.apache.org/sql/[SparkSQL] is a subproject of Spark that supports