概述

这是原始版本的,不是用phoenix的

准备HBase数据

此时 HBase的ns1下的t1是有数据的

  1. hbase(main):005:0> scan 'ns1:t1'
  2. ROW                     COLUMN+CELL
  3.  1001                   column=cf1:age, timestamp=1605086867452, value=27
  4.  1001                   column=cf1:name, timestamp=1605073387047, value=Nick
  5.  1001                   column=cf1:sex, timestamp=1605064625041, value=male
  6.  1002                   column=cf1:name, timestamp=1605073396422, value=Nick
  7.  1003                   column=cf1:name, timestamp=1605073400548, value=Nick
  8.  1004                   column=cf1:name, timestamp=1605073404158, value=Nick
  9.  1005                   column=cf1:name, timestamp=1605073406729, value=Nick
  10.  1006                   column=cf1:name, timestamp=1605073410678, value=Nick
  11.  1007                   column=cf1:name, timestamp=1605073414299, value=Nick
  12.  1008                   column=cf1:name, timestamp=1605073419160, value=Nick
  13.  1009                   column=cf1:name, timestamp=1605073422869, value=Nick
  14.  b1                     column=cf1:name, timestamp=1605160116107, value=jack
  15. 10 row(s) in 0.2440 seconds
  17. hbase(main):006:0>

pom.xml

  1.  <dependency>
  2.             <groupId>org.apache.hbase</groupId>
  3.             <artifactId>hbase-server</artifactId>
  4.             <version>1.3.1</version>
  5.         </dependency>
  7.         <dependency>
  8.             <groupId>org.apache.hbase</groupId>
  9.             <artifactId>hbase-client</artifactId>
  10.             <version>1.3.1</version>
  11.         </dependency>
  14.         <dependency>
  15.             <groupId>org.apache.spark</groupId>
  16.             <artifactId>spark-sql_2.11</artifactId>
  17.             <version>2.1.1</version>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.apache.spark</groupId>
  21.             <artifactId>spark-core_2.11</artifactId>
  22.             <version>2.1.1</version>
  23.         </dependency>

读操作

  1. package com.hbase
  3. import java.util
  5. import org.apache.hadoop.conf.Configuration
  6. import org.apache.hadoop.hbase.client.Result
  7. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  8. import org.apache.hadoop.hbase.mapreduce.TableInputFormat
  9. import org.apache.hadoop.hbase.util.Bytes
  10. import org.apache.hadoop.hbase.{Cell, CellUtil, HBaseConfiguration}
  11. import org.apache.spark.{SparkConf, SparkContext}
  12. import org.json4s.jackson.Serialization
  14. import scala.collection.mutable
  17. object HbaseRead {
  18.   def main(args: Array[String]): Unit = {
  19.     val conf: SparkConf = new SparkConf().setAppName("HbaseRead").setMaster("local[2]")
  20.     val sc: SparkContext = new SparkContext(conf)
  22.     // 连接hbase的配置
  23.     val hbaseConf: Configuration = HBaseConfiguration.create()
  25.     // zookeeper的配置
  26.     //   hbaseConf.set("hbase.zookeeper.quorum", "zjj101,zjj102,zjj103")
  27.     hbaseConf.set("hbase.zookeeper.quorum", "zjj101")
  29.     // 读取HBase的ns1下的t1表
  30.     hbaseConf.set(TableInputFormat.INPUT_TABLE, "ns1:t1")
  33.     // 从hbase读数据
  34.     val rdd1 = sc.newAPIHadoopRDD(
  35.       hbaseConf,
  36.       classOf[TableInputFormat],
  37.       classOf[ImmutableBytesWritable], // rowKey封装在这个类型中
  38.       classOf[Result]
  39.     )
  40.     // 读到数据封装下
  41.     val resultRDD = rdd1.map {
  42.       // iw只封装rowKey  result封装一行数据
  43.       case (iw, result) => {
  45.         val map = mutable.Map[String, Any]()
  46.         // 把rowKey存入到map中
  47.         map += "rowKew" -> Bytes.toString(iw.get())
  48.         // 再把每一列也存入到map中
  49.         val cells: util.List[Cell] = result.listCells()
  50.         import scala.collection.JavaConversions._ //给Java列名转成scala的.
  51.         for (cell <- cells) { // 遍历list集合
  52.           // 列名->列值
  53.           val key = Bytes.toString(CellUtil.cloneQualifier(cell))
  54.           val value = Bytes.toString(CellUtil.cloneValue(cell))
  55.           map += key -> value //  保存到map中.
  56.         }
  57.         // 为了好看把map转成json  json4s(json4scala)
  58.         implicit val df = org.json4s.DefaultFormats
  59.         Serialization.write(map)
  60.       }
  61.     }
  62.     resultRDD.collect.foreach(println)
  63.     sc.stop()
  65.   }
  66. }

运行结果

  1. {"rowKew":"1001","name":"Nick","age":"27","sex":"male"}
  2. {"rowKew":"1002","name":"Nick"}
  3. {"rowKew":"1003","name":"Nick"}
  4. {"rowKew":"1004","name":"Nick"}
  5. {"rowKew":"1005","name":"Nick"}
  6. {"rowKew":"1006","name":"Nick"}
  7. {"rowKew":"1007","name":"Nick"}
  8. {"rowKew":"1008","name":"Nick"}
  9. {"rowKew":"1009","name":"Nick"}
  10. {"rowKew":"b1","name":"jack"}

写数据

  1. package com.hbase
  3. import org.apache.hadoop.conf.Configuration
  4. import org.apache.hadoop.hbase.HBaseConfiguration
  5. import org.apache.hadoop.hbase.client.Put
  6. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  7. import org.apache.hadoop.hbase.mapreduce.{TableInputFormat, TableOutputFormat}
  8. import org.apache.hadoop.hbase.util.Bytes
  9. import org.apache.hadoop.mapreduce.Job
  10. import org.apache.spark.{SparkConf, SparkContext}
  12. object HbaseWrite {
  13.   def main(args: Array[String]): Unit = {
  14.     val conf: SparkConf = new SparkConf().setAppName("HbaseRead").setMaster("local[2]")
  15.     val sc: SparkContext = new SparkContext(conf)
  18.     val hbaseConf: Configuration = HBaseConfiguration.create()
  19.     hbaseConf.set("hbase.zookeeper.quorum", "zjj101")
  20.     //    hbaseConf.set("hbase.zookeeper.quorum", "zjj101,zjj102,zjj103")
  21.     hbaseConf.set(TableOutputFormat.OUTPUT_TABLE, "ns1:t1")
  23.     val job = Job.getInstance(hbaseConf) //得到一个Job
  24.     //设置往外写的一个类
  25.     job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])
  26.     //
  27.     job.setOutputKeyClass(classOf[ImmutableBytesWritable])
  28.     job.setOutputValueClass(classOf[Put])
  29.     //准备写的数据 第一个值是rowkey  第二个值是name ,第三个值是age
  30.     val initialRDD = sc.parallelize(
  31.       List(("1010", "zjj1", "25"),
  32.         ("1011", "zjj2", "26"),
  33.         ("1012", "zjj3", "28")))
  35.     // 先把rdd数据封装成 TableReduce需要的那种格式,才能往HBase里面写数据.
  36.     val hbaseRDD = initialRDD.map {
  37.       case (rk, name, age) =>
  38.         val rowkey = new ImmutableBytesWritable()
  39.         rowkey.set(Bytes.toBytes(rk)) //rowkey
  40.         val put = new Put(Bytes.toBytes(rk))
  41.         //cf1 是列族,
  42.         put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("name"), Bytes.toBytes(name))
  43.         put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("age"), Bytes.toBytes(age))
  44.         (rowkey, put)
  45.     }
  46.     //保存到HBase的数据库里面.
  47.     hbaseRDD.saveAsNewAPIHadoopDataset(job.getConfiguration)
  48.     sc.stop()
  49.   }
  50. }

写完查看HBase结果

仔细观察,发现rowKey为1010 ,1011, 1012 的数据已经写进HBase里面了.

  1. hbase(main):006:0> scan 'ns1:t1'
  2. ROW                     COLUMN+CELL
  3.  1001                   column=cf1:age, timestamp=1605086867452, value=27
  4.  1001                   column=cf1:name, timestamp=1605073387047, value=Nick
  5.  1001                   column=cf1:sex, timestamp=1605064625041, value=male
  6.  1002                   column=cf1:name, timestamp=1605073396422, value=Nick
  7.  1003                   column=cf1:name, timestamp=1605073400548, value=Nick
  8.  1004                   column=cf1:name, timestamp=1605073404158, value=Nick
  9.  1005                   column=cf1:name, timestamp=1605073406729, value=Nick
  10.  1006                   column=cf1:name, timestamp=1605073410678, value=Nick
  11.  1007                   column=cf1:name, timestamp=1605073414299, value=Nick
  12.  1008                   column=cf1:name, timestamp=1605073419160, value=Nick
  13.  1009                   column=cf1:name, timestamp=1605073422869, value=Nick
  14.  1010                   column=cf1:age, timestamp=1606127159307, value=25
  15.  1010                   column=cf1:name, timestamp=1606127159307, value=zjj1
  16.  1011                   column=cf1:age, timestamp=1606127159307, value=26
  17.  1011                   column=cf1:name, timestamp=1606127159307, value=zjj2
  18.  1012                   column=cf1:age, timestamp=1606127159307, value=28
  19.  1012                   column=cf1:name, timestamp=1606127159307, value=zjj3
  20.  b1                     column=cf1:name, timestamp=1605160116107, value=jack
  21. 13 row(s) in 0.4560 seconds