Spark向HBase中写入或读取数据

一时失言乱红尘 2021-11-11 12:06 412阅读 0赞

# Spark读取HBase中的数据 #

import org.apache.hadoop.conf.Configuration
    import org.apache.hadoop.hbase.{CellUtil, HBaseConfiguration}
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.hadoop.hbase.client.Result
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.spark.rdd.RDD
    
    object ReadHBaseTableDataSpark {
      def main(args: Array[String]): Unit = {
    
        var sparkConf: SparkConf = new SparkConf().setAppName("ReadHBaseTableDataSpark").
          setMaster("local[2]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          //告知那些类型需要序列化
          .registerKryoClasses(Array(classOf[ImmutableBytesWritable], classOf[Result]))
    
        //创建sparkContext对象：主要用于读取需要处理的数据，封装在RDD集合中；调度jobs执行
        val sc = new SparkContext(sparkConf)
        sc.setLogLevel("WARN")
    
        //第一步：数据的读取（输入）
        val conf: Configuration = HBaseConfiguration.create
        //设置读取的表
        conf.set(TableInputFormat.INPUT_TABLE, "ns1:orders")
        val resultRDD: RDD[(ImmutableBytesWritable, Result)] =sc.newAPIHadoopRDD(
          conf,
          classOf[TableInputFormat],         //设置数据读入格式
          classOf[ImmutableBytesWritable],   //设置输入数据key的类型
          classOf[Result]                    //设置输入数据value的类型
        )
        //第二步：数据的处理（分析）
        println(resultRDD.count())
    
        //第三步：数据的输出（输出）
    
        //取出数据
        resultRDD.take(3).foreach{case(key,result)=>{
          val rowkey = Bytes.toString(key.get())
          //输出rowkey
          println(rowkey)
          //获取每一个字段的值
          for(cell<-result.rawCells){
            val cf= Bytes.toString(CellUtil.cloneFamily(cell))
            val filed = Bytes.toString(CellUtil.cloneQualifier(cell))
            val value =Bytes.toString(CellUtil.cloneValue(cell))
            println(cf+":"+filed+":"+value)
          }
        }}
     
        //开发测试的时候，为了对每个spark app页面监控查看job的执行情况，
        //spark app运行结束4040端口就没了
    
        //关闭资源
        sc.stop()
      }
    }

# Spark向HBase中写入数据 #

import org.apache.hadoop.conf.Configuration
    import org.apache.hadoop.hbase.HBaseConfiguration
    import org.apache.hadoop.hbase.client.{Put, Result}
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable
    import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    
    object WriteDataToHBaseSpark {
    
      def main(args: Array[String]): Unit = {
        var sparkConf: SparkConf = new SparkConf().
          setAppName("WriteDataToHBaseSpark").
          setMaster("local[2]")
          //告知那些类型需要序列化
    
    
        //创建sparkContext对象：主要用于读取需要处理的数据，封装在RDD集合中；调度jobs执行
        val sc = new SparkContext(sparkConf)
        //设置日志级别
        sc.setLogLevel("WARN")
    
        //第一步：数据的读取（输入）
        val list = List(("hadoop", 324), ("spark", 9998), ("hive", 324), ("sql", 2341))
        val wordCountRDD: RDD[(String, Int)] = sc.parallelize(list,1)
    
        //转换数据格式（ImmutableBytesWritable,put）
        val putsRDD: RDD[(ImmutableBytesWritable, Put)] = wordCountRDD
          .map{
            case(word,count)=>{
              //RowKey
              val rowkey = new ImmutableBytesWritable(Bytes.toBytes(word))
              //Put对象
              val put = new Put(rowkey.get())
              put.addColumn(
                Bytes.toBytes("info"), //列簇信息
                Bytes.toBytes("count"), //字段
                Bytes.toBytes(count.toString)
              )
              (rowkey,put)
            }
          }
        var conf: Configuration = HBaseConfiguration.create
        // 设置输出OutputFormat
        conf.set("mapreduce.job.outputformat.class","org.apache.hadoop.hbase.mapreduce.TableOutputFormat")
        // 设置保存到HBase表的名称
        conf.set(TableOutputFormat.OUTPUT_TABLE, "ns1:tb_wordcount")
        // 设置输出目录
        conf.set("mapreduce.output.fileoutputformat.outputdir","/datas/spark/hbase/wc-output" + System.currentTimeMillis())
        //保存
        putsRDD.saveAsNewAPIHadoopDataset(conf)
        
        sc.stop()
      }
    
    }