Spark 下操作 HBase(1.0.0 新 API)
HBase經過七年發展,終於在今年2月底,釋出了 1.0.0 版本。這個版本提供了一些讓人激動的功能,並且,在不犧牲穩定性的前提下,引入了新的API。雖然 1.0.0 相容舊版本的 API,不過還是應該儘早地來熟悉下新版API。並且瞭解下如何與當下正紅的 Spark 結合,進行資料的寫入與讀取。鑑於國內外有關 HBase 1.0.0 新 API 的資料甚少,故作此文。
本文將分兩部分介紹,第一部分講解使用 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。
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)]
轉換成 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
提供的newAPIHadoopRDD
API將表的內容以 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) } |