学习spark操作hbase,主要是读写功能

  1. Spark操作HBase – 读操作
 1 /**
 2   * 配置hbase
 3   *
 4   * @param tableName
 5   * @param quorum
 6   * @param port
 7   * @return
 8   */
 9 def getHbaseConf(quorum: String, port: String): Configuration = {
10   // 配置hbase环境
11   val conf = HBaseConfiguration.create()
12   // 设置zookeeper地址
13   conf.set("hbase.zookeeper.quorum", quorum)
14   // 设置zookeeper端口
15   conf.set("hbase.zookeeper.property.clientPort", port)
16   // 设置读取hbase的tablename
17   conf
18 }

  

 1 def main(args: Array[String]): Unit = {
 2 
 3   // 获取spark环境 设置local使程序在本地运行,不需要安装Spark集群 这里使用的是yarn模式
 4   val sparkConf = new SparkConf().setAppName("Spark-Hbase-Read").setMaster("local[2]")
 5   val sc = new SparkContext(sparkConf)
 6   // 获取hbase相关配置信息
 7   val hbaseConf = HbaseUtils.getHbaseConf("centos01,centos02,centos03", "2181")
 8   hbaseConf.set(TableInputFormat.INPUT_TABLE, "movie_wordcount")
 9   // 将 hbase 读取信息转换成rdd
10   val hbaseRdd = sc.newAPIHadoopRDD(hbaseConf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result]).cache()
11 
12   hbaseRdd.map(x => {
13     val result = x._2
14     val row = Bytes.toString(result.getRow)
15     val word = Bytes.toString(result.getValue("word".getBytes(), "word".getBytes()))
16     // 单词量是int 所以需要用 Bytes.toInt 否则会出现乱码
17     val count = Bytes.toInt(result.getValue("word".getBytes(), "count".getBytes()))
18 
19     println(row, word, count)
20     (row, word, count)
21   }).saveAsTextFile("/wordcount/output3")
22 
23 
24 }

打包提交到集群,运行 命令:./bin/spark-submit --class com.xxx.xx.scala.hbase.SparkHbaseR ./localjar/sc-1.0-SNAPSHOT-jar-with-dependencies.jar 

  saveAsTextFile 默认保存地址是hdfs上的,所以去hdfs上查看结果,结果是一个文件夹,

  查看命令: hadoop fs -ls /wordcount/output3

  

  2.Spark操作HBase – 写操作

 

object SparkHbaseW {

def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("Spark-Hbase-Write").setMaster("yarn")
val sc = new SparkContext(conf)
val file = sc.textFile("/spark/movie/wordcount/pinglun.txt")
// 使用中文分词
val rdd = file.flatMap(line => {
getWords(line, filter(new Array[String](0)))
}).map(x => (x, 1)).reduceByKey(_ + _)
// 保存一份结果到hdfs
rdd.saveAsTextFile(args(0))

// 插入方式一
rdd.foreachPartition(x => {
x.foreach(y => {
// 将数组插入hbase
val table = HbaseUtils.getTable(HbaseUtils.getHbaseConf("centos01,centos02,centos03", "2181"), "movie_wordcount")
val family = Bytes.toBytes("word")
val wordColum = Bytes.toBytes("word")
val countColum = Bytes.toBytes("count")

val uuid = UUID.randomUUID()
val wordPut = new Put(Bytes.toBytes(uuid.toString))
wordPut.addColumn(family, wordColum, Bytes.toBytes(y._1))
wordPut.addColumn(family, countColum, Bytes.toBytes(y._2))
table.put(wordPut)
})
})

// 插入方式二 (批量插入)
// rdd.foreachPartition(x => {
// val table = HbaseUtils.getTable(HbaseUtils.getHbaseConf("centos01,centos02,centos03", "2181"), "movie_wordcount")
// val puts = new java.util.LinkedList[Put]()
// x.foreach(y => {
// // 将数组插入hbase
// val family = Bytes.toBytes("word")
// val wordColum = Bytes.toBytes("word")
// val countColum = Bytes.toBytes("count")
// val uuid = UUID.randomUUID()
// val wordPut = new Put(Bytes.toBytes(uuid.toString))
// wordPut.addColumn(family, wordColum, Bytes.toBytes(y._1))
// wordPut.addColumn(family, countColum, Bytes.toBytes(y._2))
// puts.add(wordPut)
// })
// table.put(puts)
// })


}


/**
* 分词停止符
*
* @param stopWords
* @return
*/
def filter(stopWords: Array[String]): StopRecognition = {
// add stop words
val filter = new StopRecognition
filter.insertStopNatures("w") // filter punctuation
filter.insertStopNatures("m") // filter m pattern
filter.insertStopNatures("null") // filter null
filter.insertStopNatures("<br />") // filter <br />
filter.insertStopRegexes("^[a-zA-Z]{1,}") //filter English alphabet
filter.insertStopRegexes("^[0-9]+") //filter number
filter.insertStopRegexes("[^a-zA-Z0-9\\u4e00-\\u9fa5]+")
filter.insertStopRegexes("\t")
for (x <- stopWords) {
filter.insertStopWords(x)
}
filter
}

/**
* 分词统计
*
* @param text
* @param filter
* @return
*/
def getWords(text: String, filter: StopRecognition): ArrayBuffer[String] = {
val words = new mutable.ArrayBuffer[String]()
val terms = ToAnalysis.parse(text).recognition(filter).getTerms
for (i <- 0 until terms.size()) {
val word = terms.get(i).getName
if (word.length >= 0) {
words += word
}
}
words
}


}

 

版权声明:本文为chengzhihua原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://www.cnblogs.com/chengzhihua/p/11428673.html