SparkSQL / DataFrames
HBase-Spark连接器（在HBase-Spark模块中）利用Spark-1.2.0中引入的DataSource API （SPARK-3247），弥补了简单HBase KV存储和复杂关系SQL查询之间的差距，使用户能够使用Spark在HBase上执行复杂的数据分析工作。HBase Dataframe是标准的Spark Dataframe，能够与任何其他数据源（如Hive，Orc，Parquet，JSON等）进行交互。HBase-Spark Connector应用关键技术，如分区修剪，列修剪，谓词叠加和数据局部性。
要使用HBase-Spark连接器，用户需要为HBase和Spark表之间的模式映射定义Catalog，准备数据并填充HBase表，然后加载HBase DataFrame。之后，用户可以使用SQL查询在HBase表中进行集成查询和访问记录。以下说明了基本程序。
定义目录
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
Catalog定义了HBase和Spark表之间的映射。该目录有两个关键部分。一个是rowkey定义，另一个是Spark中的表列与HBase中的列族和列限定符之间的映射。上面定义了一个HBase表的模式，其名称为table1，行键为key，列数为col1 -col8。请注意，还必须将rowkey详细定义为column (col0)，该列具有特定的cf（rowkey）。
保存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是一个本地Scala集合，它有256个HBaseRecord对象。 sc.parallelize(data)函数分配data以形成RDD。toDF返回一个DataFrame。 writefunction返回一个DataFrameWriter，它用于将DataFrame写入外部存储系统（例如，HBase）。给定具有指定模式的DataFrame catalog，save函数将创建一个包含5个区域的HBase表，并将DataFrame保存在其中。
加载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中处理结构化数据（行和列）的入口点。 read返回一个DataFrameReader，可用于以DataFrame的形式读取数据。 option函数将基础数据源的输入选项添加到DataFrameReader，format函数指定DataFrameReader的输入数据源格式。该load()函数将输入作为DataFrame加载。withCatalog函数返回的日期框df可用于访问HBase表。
语言综合查询
val s = df.filter(($"col0" <= "row050" && $"col0" > "row040") ||
$"col0" === "row005" ||
$"col0" <= "row005")
.select("col0", "col1", "col4")
s.show
DataFrame可以执行各种操作，例如join，sort，select，filter，orderBy等。上面的df.filter使用给定的SQL表达式过滤行。select选择一组列： col0，col1和col4。
SQL查询
df.registerTempTable("table1")
sqlContext.sql("select count(col1) from table1").show
registerTempTabledf使用表名将DataFrame注册为临时表table1。此临时表的生命周期与用于创建df的SQLContext相关联。sqlContext.sql函数允许用户执行SQL查询。
其他
示例-使用不同时间戳的查询
在HBaseSparkConf中，可以设置与时间戳相关的四个参数。它们分别是TIMESTAMP，MIN_TIMESTAMP，MAX_TIMESTAMP和MAX_VERSIONS。用户可以使用MIN_TIMESTAMP和MAX_TIMESTAMP查询具有不同时间戳或时间范围的记录。与此同时，在下面的示例中使用具体值而不是tsSpecified和oldMs。
下面的示例显示了如何使用不同的时间戳加载df DataFrame。tsSpecified由用户指定。HBaseTableCatalog定义HBase和Relation关系模式。writeCatalog定义模式映射的目录。
val df = sqlContext.read
.options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseSparkConf.TIMESTAMP -> tsSpecified.toString))
.format("org.apache.hadoop.hbase.spark")
.load()
下面的示例显示了如何加载具有不同时间范围的df DataFrame。oldMs由用户指定。
val df = sqlContext.read
.options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseSparkConf.MIN_TIMESTAMP -> "0",
HBaseSparkConf.MAX_TIMESTAMP -> oldMs.toString))
.format("org.apache.hadoop.hbase.spark")
.load()
加载df DataFrame后，用户可以查询数据。
df.registerTempTable("table")
sqlContext.sql("select count(col1) from table").show
示例-本地Avro支持
HBase-Spark Connector支持不同的数据格式，如Avro，Jason等。下面的用例显示了spark是如何支持Avro的。用户可以直接将Avro记录保存到HBase中。在内部，Avro模式自动转换为本机Spark Catalyst数据类型。请注意，HBase表中的两个键值部分都可以用Avro格式定义。
1）定义模式映射的目录：
def catalog = s"""{
|"table":{"namespace":"default", "name":"Avrotable"},
|"rowkey":"key",
|"columns":{
|"col0":{"cf":"rowkey", "col":"key", "type":"string"},
|"col1":{"cf":"cf1", "col":"col1", "type":"binary"}
|}
|}""".stripMargin
catalog是名为Avrotable的HBase表的模式。行键作为键和一列col1。还必须将rowkey详细定义为column (col0)，该列具有特定的cf（rowkey）。
2）准备数据：
object AvroHBaseRecord {
val schemaString =
s"""{"namespace": "example.avro",
| "type": "record", "name": "User",
| "fields": [
| {"name": "name", "type": "string"},
| {"name": "favorite_number", "type": ["int", "null"]},
| {"name": "favorite_color", "type": ["string", "null"]},
| {"name": "favorite_array", "type": {"type": "array", "items": "string"}},
| {"name": "favorite_map", "type": {"type": "map", "values": "int"}}
| ] }""".stripMargin
val avroSchema: Schema = {
val p = new Schema.Parser
p.parse(schemaString)
}
def apply(i: Int): AvroHBaseRecord = {
val user = new GenericData.Record(avroSchema);
user.put("name", s"name${"%03d".format(i)}")
user.put("favorite_number", i)
user.put("favorite_color", s"color${"%03d".format(i)}")
val favoriteArray = new GenericData.Array[String](2, avroSchema.getField("favorite_array").schema())
favoriteArray.add(s"number${i}")
favoriteArray.add(s"number${i+1}")
user.put("favorite_array", favoriteArray)
import collection.JavaConverters._
val favoriteMap = Map[String, Int](("key1" -> i), ("key2" -> (i+1))).asJava
user.put("favorite_map", favoriteMap)
val avroByte = AvroSedes.serialize(user, avroSchema)
AvroHBaseRecord(s"name${"%03d".format(i)}", avroByte)
}
}
val data = (0 to 255).map { i =>
AvroHBaseRecord(i)
}
首先定义schemaString，然后解析得到avroSchema。avroSchema用于生成AvroHBaseRecord。用户准备的data是一个包含256个AvroHBaseRecord对象的本地Scala集合。
3）保存DataFrame：
sc.parallelize(data).toDF.write.options(
Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "5"))
.format("org.apache.spark.sql.execution.datasources.hbase")
.save()
给定具有指定模式的数据框catalog，上面将创建一个包含5个区域的HBase表，并将数据框保存在其中。
4）加载DataFrame
def avroCatalog = s"""{
|"table":{"namespace":"default", "name":"avrotable"},
|"rowkey":"key",
|"columns":{
|"col0":{"cf":"rowkey", "col":"key", "type":"string"},
|"col1":{"cf":"cf1", "col":"col1", "avro":"avroSchema"}
|}
|}""".stripMargin
def withCatalog(cat: String): DataFrame = {
sqlContext
.read
.options(Map("avroSchema" -> AvroHBaseRecord.schemaString, HBaseTableCatalog.tableCatalog -> avroCatalog))
.format("org.apache.spark.sql.execution.datasources.hbase")
.load()
}
val df = withCatalog(catalog)
在withCatalog函数中，read返回一个DataFrameReader，可用于以DataFrame的形式读取数据。该option函数将基础数据源的输入选项添加到DataFrameReader。有两个选项：一个是设置avroSchema为AvroHBaseRecord.schemaString，一个是设置HBaseTableCatalog.tableCatalog为avroCatalog。该load()函数将输入作为DataFrame加载。withCatalog函数返回的日期框df可用于访问HBase表。
5）SQL查询：
df.registerTempTable("avrotable")
val c = sqlContext.sql("select count(1) from avrotable").
加载df DataFrame后，用户可以查询数据。registerTempTable使用表名avrotable将df DataFrame注册为临时表。sqlContext.sql函数允许用户执行SQL查询。
文章永久链接：https://ddkk.com/?p=12384