HBase經(jīng)過七年發(fā)展，終于在今年2月底，發(fā)布了 1.0.0 版本。這個(gè)版本提供了一些讓人激動(dòng)的功能，并且，在不犧牲穩(wěn)定性的前提下，引入了新的API。雖然 1.0.0 兼容舊版本的 API，不過還是應(yīng)該盡早地來熟悉下新版API。并且了解下如何與當(dāng)下正紅的
Spark 結(jié)合，進(jìn)行數(shù)據(jù)的寫入與讀取。鑒于國內(nèi)外有關(guān) HBase 1.0.0 新 API 的資料甚少，故作此文。
本文將分兩部分介紹，第一部分講解使用 HBase 新版 API 進(jìn)行 CRUD 基本操作；第二部分講解如何將 Spark 內(nèi)的 RDDs 寫入 HBase 的表中，反之，HBase 中的表又是如何以 RDDs 形式加載進(jìn) Spark 內(nèi)的。
環(huán)境配置
為了避免版本不一致帶來不必要的麻煩，API 和 HBase環(huán)境都是 1.0.0 版本。HBase 為單機(jī)模式，分布式模式的使用方法類似，只需要修改HBaseConfiguration的配置即可。
開發(fā)環(huán)境中使用 SBT 加載依賴項(xiàng)
name := "SparkLearn"
version := "1.0"
scalaVersion := "2.10.4"
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.3.0"
libraryDependencies += "org.apache.hbase" % "hbase-client" % "1.0.0"
libraryDependencies += "org.apache.hbase" % "hbase-common" % "1.0.0"
libraryDependencies += "org.apache.hbase" % "hbase-server" % "1.0.0"
HBase 的 CRUD 操作
新版 API 中加入了 Connection，HAdmin成了Admin，HTable成了Table，而Admin和Table只能通過Connection獲得。Connection的創(chuàng)建是個(gè)重量級(jí)的操作，由于Connection是線程安全的，所以推薦使用單例，其工廠方法需要一個(gè)HBaseConfiguration。
val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("hbase.zookeeper.quorum", "master")
//Connection 的創(chuàng)建是個(gè)重量級(jí)的工作，線程安全，是操作hbase的入口
val conn = ConnectionFactory.createConnection(conf)
創(chuàng)建表
使用Admin創(chuàng)建和刪除表
val userTable = TableName.valueOf("user")
//創(chuàng)建 user 表
val tableDescr = new HTableDescriptor(userTable)
tableDescr.addFamily(new HColumnDescriptor("basic".getBytes))
println("Creating table `user`. ")
if (admin.tableExists(userTable)) {
admin.disableTable(userTable)
admin.deleteTable(userTable)
}
admin.createTable(tableDescr)
println("Done!")
插入、查詢、掃描、刪除操作
HBase 上的操作都需要先創(chuàng)建一個(gè)操作對(duì)象Put,Get,Delete等，然后調(diào)用Table上的相對(duì)應(yīng)的方法
try{
//獲取 user 表
val table = conn.getTable(userTable)
try{
//準(zhǔn)備插入一條 key 為 id001 的數(shù)據(jù)
val p = new Put("id001".getBytes)
//為put操作指定 column 和 value （以前的 put.add 方法被棄用了）
p.addColumn("basic".getBytes,"name".getBytes, "wuchong".getBytes)
//提交
table.put(p)
//查詢某條數(shù)據(jù)
val g = new Get("id001".getBytes)
val result = table.get(g)
val value = Bytes.toString(result.getValue("basic".getBytes,"name".getBytes))
println("GET id001 :"+value)
//掃描數(shù)據(jù)
val s = new Scan()
s.addColumn("basic".getBytes,"name".getBytes)
val scanner = table.getScanner(s)
try{
for(r <- scanner){
println("Found row: "+r)
println("Found value: "+Bytes.toString(
r.getValue("basic".getBytes,"name".getBytes)))
}
}finally {
//確保scanner關(guān)閉
scanner.close()
}
//刪除某條數(shù)據(jù),操作方式與 Put 類似
val d = new Delete("id001".getBytes)
d.addColumn("basic".getBytes,"name".getBytes)
table.delete(d)
}finally {
if(table != null) table.close()
}
}finally {
conn.close()
}
Spark 操作 HBase
寫入 HBase
首先要向 HBase 寫入數(shù)據(jù)，我們需要用到PairRDDFunctions.saveAsHadoopDataset。因?yàn)?HBase 不是一個(gè)文件系統(tǒng)，所以saveAsHadoopFile方法沒用。
def saveAsHadoopDataset(conf: JobConf): Unit
Output the RDD to any
Hadoop-supported storage system, using a Hadoop JobConf object for that storage system
這個(gè)方法需要一個(gè) JobConf 作為參數(shù)，類似于一個(gè)配置項(xiàng)，主要需要指定輸出的格式和輸出的表名。
Step 1：我們需要先創(chuàng)建一個(gè) JobConf。
import org.apache.hadoop.hbase.mapred.TableOutputFormat//定義 HBase 的配置
val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("hbase.zookeeper.quorum", "master")
//指定輸出格式和輸出表名
val jobConf = new JobConf(conf,this.getClass)
jobConf.setOutputFormat(classOf[TableOutputFormat])
jobConf.set(TableOutputFormat.OUTPUT_TABLE,"user")
Step 2： RDD 到表模式的映射
在 HBase 中的表 schema 一般是這樣的：
row cf:col_1 cf:col_2
而在Spark中，我們操作的是RDD元組，比如(1,"lilei",14), (2,"hanmei",18)。我們需要將RDD[(uid:Int, name:String, age:Int)] 轉(zhuǎn)換成RDD[(ImmutableBytesWritable, Put)]。所以，我們定義一個(gè) convert 函數(shù)做這個(gè)轉(zhuǎn)換工作
def convert(triple: (Int, String, Int)) = {
val p = new Put(Bytes.toBytes(triple._1))
p.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("name"),Bytes.toBytes(triple._2))
p.addColumn(Bytes.toBytes("basic"),Bytes.toBytes("age"),Bytes.toBytes(triple._3))
(new ImmutableBytesWritable, p)
}
Step 3： 讀取RDD并轉(zhuǎn)換
//read RDD data from somewhere and convert
val rawData = List((1,"lilei",14), (2,"hanmei",18), (3,"someone",38))
val localData = sc.parallelize(rawData).map(convert)
Step 4： 使用saveAsHadoopDataset方法寫入HBase
localData.saveAsHadoopDataset(jobConf)
讀取 HBase
Spark讀取HBase，我們主要使用SparkContext 提供的newAPIHadoopRDDAPI將表的內(nèi)容以 RDDs 的形式加載到 Spark 中。
val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("hbase.zookeeper.quorum", "master")
//設(shè)置查詢的表名
conf.set(TableInputFormat.INPUT_TABLE, "user")
val usersRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
classOf[org.apache.hadoop.hbase.client.Result])
val count = usersRDD.count()
println("Users RDD Count:" + count)
usersRDD.cache()
//遍歷輸出
usersRDD.foreach{ case (_,result) =>
val key = Bytes.toInt(result.getRow)
val name = Bytes.toString(result.getValue("basic".getBytes,"name".getBytes))
val age = Bytes.toInt(result.getValue("basic".getBytes,"age".getBytes))
println("Row key:"+key+" Name:"+name+" Age:"+age)
}華為開源項(xiàng)目 spark-sql-on-hbase :
https://github.com/HuaweiBigData/astro
