Spark本地模式下使用HBase进行MapReduce操作

package test

import _root_.util.FileUtil
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, KeyValue}
import org.apache.hadoop.conf.Configuration
import org.apache.spark.api.java.JavaPairRDD
import org.apache.spark.{SparkConf, SparkContext}
import option.GenarateOption
class hello(year:String){
  def doSpark(): java.lang.Integer ={
    hello.run(year)
  }
}
object hello {
  val TABLE_NAME = "expense_list"                         //hbase表名
  val TABLE_COLUMN_FAIMLY = Bytes.toBytes("expense_info") //列族
  val TABLE_COLUMN_DATE = Bytes.toBytes("apply_date")     //时间
  val TABLE_COLUMN_TYPE = Bytes.toBytes("expense_type")   //类型
  val TABLE_COLUMN_SUM = Bytes.toBytes("fee_total")        //总额
  val YEAR_PATH = Thread.currentThread.getContextClassLoader.getResource("").getFile + "//line.tmp"
  def run(year:String): java.lang.Integer ={
     val conf = new SparkConf().setAppName("test")
    conf.setMaster("local")//本地模式运行spark
    val sc:SparkContext = new SparkContext(conf)
    try{
    deleteOutPath()//清空输出目录
    val hbaseConfig = HBaseConfiguration.create()//hbase的配置

    hbaseConfig.set("hbase.zookeeper.quorum","cdh.codsway.com")//设置zookeeper主机
    hbaseConfig.set(TableInputFormat.INPUT_TABLE,TABLE_NAME)//访问的表名
    val maps = getResult(sc,hbaseConfig,year) //将hbase读取的数据转换为Map[String,List[Double]]的形式
    if(maps==null||maps.size()<=0){
      return 0       //没有数据时返回0
    }
    val line_json = GenarateOption.generateLineByMonth(maps,year)//构造echarts所需的option结构
    FileUtil.write(YEAR_PATH,line_json)//写入文件
    return 1   //有数据返回1
    }catch {
      case e:Exception => return -1 //程序出现异常则返回-1
    }finally {
      sc.stop() //关闭spark连接
    }
  }
  def getResult(sc: SparkContext,hbaseConfig: Configuration,year:String): java.util.HashMap[String,java.util.LinkedList[java.lang.Double]] ={
    val map = new java.util.HashMap[String,java.util.LinkedList[java.lang.Double]]()
    //spark RDD
     val hbaseRDD = sc.newAPIHadoopRDD(hbaseConfig,classOf[TableInputFormat],classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result])
    val keyValues = hbaseRDD.map(tuple =>tuple._2).map(result =>result.raw())//获数据取
    val rdd = keyValues.filter(line =>filterByDate(line,year)).map(line => toKeyValue(line))//将数据重新拼接成key-value形式
    val pair = JavaPairRDD.fromJavaRDD(rdd)//将rdd转成pairRDD （rdd格式必须为key-value）pairRDD才能进行reduceByKey等操作
    val list =  pair.groupByKey().collect()//按key 进行groupBy操作
    val it = list.iterator()
    var key = ""
    var mapCount = 0;
    while(it.hasNext){
      var count:Double = 0
      val next = it.next()