spark 读取hbase数据并转化为dataFrame

￡神魔★判官ぃ 2022-07-21 01:52 303阅读 0赞

最近两天研究spark直接读取hbase数据，并将其转化为dataframe。之所以这么做，

1、公司的数据主要存储在hbase之中

2、使用dataframe，更容易实现计算。

尽管hbase的官方已经提供了hbase-spark 接口，但是并未对外发布，而且目前的项目又有这方面的需求，且网上关于这么方面的参考比较少，

故贴出来，代码如下，仅供参考

import org.apache.hadoop.hbase.client._
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.hadoop.hbase.{
      TableName, HBaseConfiguration}
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.spark.sql.SQLContext
    import org.apache.spark.{SparkContext, SparkConf}
    
    /** * Created by seagle on 6/28/16. */ object HBaseSpark {
      def main(args:Array[String]): Unit ={
    
        // 本地模式运行,便于测试 val sparkConf = new SparkConf().setMaster("local").setAppName("HBaseTest")
    
        // 创建hbase configuration val hBaseConf = HBaseConfiguration.create()
        hBaseConf.set(TableInputFormat.INPUT_TABLE,"bmp_ali_customer")
    
        // 创建 spark context val sc = new SparkContext(sparkConf)
        val sqlContext = new SQLContext(sc)
        import sqlContext.implicits._
    
        // 从数据源获取数据 val hbaseRDD = sc.newAPIHadoopRDD(hBaseConf,classOf[TableInputFormat],classOf[ImmutableBytesWritable],classOf[Result])
    
        // 将数据映射为表 也就是将 RDD转化为 dataframe schema val shop = hbaseRDD.map(r=>(
          Bytes.toString(r._2.getValue(Bytes.toBytes("info"),Bytes.toBytes("customer_id"))),
          Bytes.toString(r._2.getValue(Bytes.toBytes("info"),Bytes.toBytes("create_id")))
          )).toDF("customer_id","create_id")
    
        shop.registerTempTable("shop")
    
        // 测试 val df2 = sqlContext.sql("SELECT customer_id FROM shop")
    
        df2.foreach(println)
      }
    
    }

代码能够运行的前提是

1、 引用了 spark-sql  jar

2、配置了Hbase-site.xml ，并将其放在工程的根目录下