本文將分兩部分介紹，第一部分講解使用 HBase 新版 API 進行 CRUD 基本操作；第二部分講解如何將 Spark 內的 RDDs 寫入 HBase 的表中，反之，HBase 中的表又是如何以 RDDs 形式加載進 Spark 內的。

環境配置

為了避免版本不一致帶來不必要的麻煩，API 和 HBase環境都是 1.0.0 版本。HBase 為單機模式，分布式模式的使用方法類似，只需要修改HBaseConfiguration的配置即可。

開發環境中使用 SBT 加載依賴項

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的創建是個重量級的操作，由於Connection是線程安全的，所以推薦使用單例，其工廠方法需要一個HBaseConfiguration。

val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("hbase.zookeeper.quorum", "master")

//Connection 的創建是個重量級的工作，線程安全，是操作hbase的入口
val conn = ConnectionFactory.createConnection(conf)

 

創建表

使用Admin創建和刪除表

val userTable = TableName.valueOf("user")

//創建 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 上的操作都需要先創建一個操作對象Put,Get,Delete等，然后調用Table上的相對應的方法

try{
  //獲取 user 表
  val table = conn.getTable(userTable)

  try{
    //准備插入一條 key 為 id001 的數據
    val p = new Put("id001".getBytes)
    //為put操作指定 column 和 value （以前的 put.add 方法被棄用了）
    p.addColumn("basic".getBytes,"name".getBytes, "wuchong".getBytes)
    //提交
    table.put(p)

    //查詢某條數據
    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)

    //掃描數據
    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關閉
      scanner.close()
    }

    //刪除某條數據,操作方式與 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 寫入數據，我們需要用到PairRDDFunctions.saveAsHadoopDataset。因為 HBase 不是一個文件系統，所以saveAsHadoopFile方法沒用。

def saveAsHadoopDataset(conf: JobConf): Unit
Output the RDD to any Hadoop-supported storage system, using a Hadoop JobConf object for that storage system

這個方法需要一個 JobConf 作為參數，類似於一個配置項，主要需要指定輸出的格式和輸出的表名。

Step 1：我們需要先創建一個 JobConf。

//定義 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)] 轉換成 RDD[(ImmutableBytesWritable, Put)]。所以，我們定義一個 convert 函數做這個轉換工作

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並轉換

//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將表的內容以 RDDs 的形式加載到 Spark 中。

val conf = HBaseConfiguration.create()
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("hbase.zookeeper.quorum", "master")

//設置查詢的表名
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)
}