HBase-Spark Connector(在HBase-Spark 模块中）利用了在Spark-1.2.0中引入的DataSource API(SPARK-3247)，在简单的HBase KV存储和复杂的关系型SQL查询之间架起了桥梁，使得用户可以在HBase上使用Spark执行复杂的数据分析工作。HBase Dataframe是一个标准的Spark Dataframe，能够与任何其他的数据源进行交互，比如Hive,Orc，Parquet,JSON等。HBase-Spark Connector应用了关键技术，如分区剪枝（partition
pruning)，列剪枝(column pruning)，谓詞下推(predicate pushdown)和数据局部性（data locality）。
要使用HBase-Spark Connector，用户需要定义在HBase和Spark表之间的映射关系的schema目录，准备数据，并且填充到HBase表中，然后加载HBase Dataframe。之后，用户可以使用SQL查询做集成查询和访问记录HBase的表。以下描述了这个的基本步骤：
１、定义目录（Define catalog）
２、保存DataFrame
３、加载DataFrame
４、SQL 查询
1.
Define catalog
def catalog = s"""{
|"table":{"namespace":"default", "name":"table1"},
|"rowkey":"key",
|"columns":{
|"col0":{"cf":"rowkey", "col":"key", "type":"string"},
|"col1":{"cf":"cf1", "col":"col1", "type":"boolean"},
|"col2":{"cf":"cf2", "col":"col2", "type":"double"},
|"col3":{"cf":"cf3", "col":"col3", "type":"float"},
|"col4":{"cf":"cf4", "col":"col4", "type":"int"},
|"col5":{"cf":"cf5", "col":"col5", "type":"bigint"},
|"col6":{"cf":"cf6", "col":"col6", "type":"smallint"},
|"col7":{"cf":"cf7", "col":"col7", "type":"string"},
|"col8":{"cf":"cf8", "col":"col8", "type":"tinyint"}
|}
|}""".stripMargin
目录定义了HBase和Spark表之间的映射。这个目录有两个关键部分。一个是rowkey定义，另一个是Spark表中列与HBase中的列族和列限定符之间的映射。上面定义了名称为table1，行健为key和列数（col1
- col8）的HBase表的模式。请注意，rowkey还必须作为列（col0）进行详细定义，该列具有特定的cf（rowkey）。2.
Save the DataFrame
case class HBaseRecord(
col0: String,
col1: Boolean,
col2: Double,
col3: Float,
col4: Int,
col5: Long,
col6: Short,
col7: String,
col8: Byte)
object HBaseRecord
{
def apply(i: Int, t: String): HBaseRecord = {
val s = s"""row${"%03d".format(i)}"""
HBaseRecord(s,
i % 2 == 0,
i.toDouble,
i.toFloat,
i,
i.toLong,
i.toShort,
s"String$i: $t",
i.toByte)
}
}
val data = (0 to 255).map { i => HBaseRecord(i, "extra")}
sc.parallelize(data).toDF.write.options(
Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "5"))
.format("org.apache.hadoop.hbase.spark ")
.save()
由用户准备的data是具有256个HBaseRecord对象的本地Scala集合。sc.parallelize（data）函数分配data以形成RDD。toDF
返回一个DataFrame。write函数返回一个用于将DataFrame写入外部存储系统的DataFrameWriter(例如 HBase)。给定一个具有指定模式CataLog的DataFrame，save函数将创建一个包含5个区域的HBase表，并将DataFrame保存在里面。
3.
Load the DataFrame
def withCatalog(cat: String): DataFrame = {
sqlContext
.read
.options(Map(HBaseTableCatalog.tableCatalog->cat))
.format("org.apache.hadoop.hbase.spark")
.load()
}
val df = withCatalog(catalog)
在'withCatalog'函数中，sqlContext是一个SQLContext变量，它是在Spark中处理结构化数据（行和列）的入口点。读取返回一个DataFrameReader，它可以用来读取DataFrame中的数据。option函数将底层数据源的输入选项添加到DataFrameReader，format函数指定DataFrameReader的输入数据源格式。
load（）函数以DataFrame的形式加载输入。withCatalog函数返回的data frame
df可用于访问HBase表，如4.4和4.5。
4.
Language Integrated Query
val s = df.filter(($"col0" <= "row050" && $"col0" > "row040") ||
$"col0" === "row005" ||
$"col0" <= "row005")
.select("col0", "col1", "col4")
s.show
DataFrame可以执行各种操作，如连接，排序，选择，过滤，orderBy等等。上面的df.filter使用给定的SQL表达式过滤行。
select选择一组列：col0，col1和col4。
5.
SQL Query
df.registerTempTable("table1")
sqlContext.sql("select count(col1) from table1").show
registerTempTable使用表名table1将df DataFrame注册为临时表。此临时表的生命周期与用于创建df的SQLContext绑定。sqlContext.sql函数允许用户执行SQL查询
参考资料：https://hbase.apache.org/book.html#_sparksql_dataframes
这里面还有一个通过时间查询的例子。感兴趣的朋友可以看看去。