spark rdd 读写hbase数据

た入场券 2022-04-13 02:25 252阅读 0赞

hive 作为mapreduce计算引擎， 可以使用hql来操作hbase表 （hbase也可以访问hive表的数据)， spark也是计算引擎，按理也是可以读写hbase数据的

使用idea + maven 读写hbase数据， 操作如下：

# 1， 添加maven依赖 #

<dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-client</artifactId>
                <version>1.2.6</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-server</artifactId>
                <version>1.2.6</version>
            </dependency>
    
    	<dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.11</artifactId>
                <version>2.1.0</version>
            </dependency>

# 2， spark读hbase表 #

import org.apache.hadoop.hbase.HBaseConfiguration
    import org.apache.hadoop.hbase.client.Result
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.spark._
    
    /** * Created by wang on 18-11-26. */
    object Get { 
      def main(args: Array[String]): Unit = { 
    
    
        val sc_conf=new SparkConf().setMaster("local[2]").setAppName("读取Hbase中的数据")
        val sc=new SparkContext(sc_conf)
    
        //设置查询的表名
        val conf=HBaseConfiguration.create()
        conf.set(TableInputFormat.INPUT_TABLE,"ns1:person")
        val RDD=sc.newAPIHadoopRDD(conf,classOf[TableInputFormat],classOf[ImmutableBytesWritable],classOf[Result])
       RDD.cache()
    
        val count=RDD.count()
        println("Students RDD Count:"+count)
    
        //遍历输出
        RDD.foreach({ case (_,result)=>
          val key=Bytes.toString(result.getRow)
          val id=Bytes.toString(result.getValue("f1".getBytes,"id".getBytes))
          val name=Bytes.toString(result.getValue("f1".getBytes,"name".getBytes))
          val age=Bytes.toString(result.getValue("f1".getBytes,"age".getBytes))
    
          println("ROW:"+key+" id: "+id +" name: " +name+" Age: "+age)
        })
        
    // RDD.foreachPartition(iter => { 
    // while (iter.hasNext) { 
    // val tup = iter.next()
    // val result = tup._2
    //
    // //遍历输出
    // // println(key + ", CREATETIME= " + CREATETIME ......
    // }
    // })//foreachPartiton
      }
    }

# 3， spark写hbase表 #

import org.apache.hadoop.hbase.HBaseConfiguration
    import org.apache.hadoop.hbase.client.{ HTable, Put}
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object Write2 { 
      def main(args: Array[String]): Unit = { 
      
        val sparkConf = new SparkConf().setAppName("HBaseTest").setMaster("local[*]")
        sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        val sc = new SparkContext(sparkConf)
    
        //创建rdd
        val rdd = sc.makeRDD(1 to 10)
    
        //写入数据到hbase
        rdd.foreach(x => { 
    
          //创建hbase连接
          val hbaseConf = HBaseConfiguration.create()
          hbaseConf.set("hbase.zookeeper.quorum", "localhost")
          hbaseConf.set("hbase.zookeeper.property.clientPort", "2181")
    
          //表
          val t1 = new HTable(hbaseConf, "t1")
          t1.setAutoFlush(false)
    
          //写入数据
          val put = new Put(Bytes.toBytes("row" + x))
          put.add(Bytes.toBytes("f1"), Bytes.toBytes("num"), Bytes.toBytes(x))
    
          t1.put(put)
          t1.flushCommits
        })
    
        //关闭spark
        sc.stop()
      }
    }