devbox
小惠珠哦
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spark读写Hbase（用户画像）

作者：小惠珠哦 | 2024-07-24 13:43:45

赞

踩

spark读写hbase

背景

依旧是公司用户画像项目，目前方案是将hive聚合之后的标签表全部倒入mysql，然后在ES建立索引，虽然限定了最大查询范围为90天的数据，但是面对千万级的用户量，90天的数据依旧是非常庞大，每天查询的效率依旧是在30分钟以上，所以准备对这块进行优化。

在公司层面进行调研之后发现，公司遗留了一个小的Hbase集群，集群配置：

1 active master, 1 backup masters, 2 servers。

但是问题是集群的版本为：1.1.2，非常低，在上篇文章中：

浅谈Hbase在用户画像上的应用

为了做用户画像的流程打通，本地建立的Hbase版本为1.3.6，spark版本为2.4，所以整套体系都不支持公司原来的hbase集群体系，为了保障之后的用户画像能落地，有两套解决方案：

1、公司层面的Hbase集群升级，由于历史包袱太重，之前的版本虽然老，但是依旧有部分的数据在上面跑，如果版本升级，后续对应的下游系统中间件可能会出现不兼容的问题，而且在Hbase层做适配需要调研太多的下游业务的使用场景，成本太高，所以未选用。

2、将本地的伪分布式Hbase进行降级，同时spark版本也进行降级处理。目前测试环境选定的Hbase版本为1.1.2，spark版本为2.1.1。

下面的文章主要是基于这两个版本的中间件，进行spark对Hbase的读写操作。

介绍

1、Hbase版本降级，1.1.2版本为2015年的版本

wget archive.apache.org/dist

2、按照上篇文章对Hbase进行配置和按照测试

访问页面：localhost:60010/master-

启动hbase shell，并插入数据：

Scan的结果：Row_key、列族、列名、时间戳、value值

实际结果用图标展现为：

转换为常见的关系型数据库的视角来看的话：

从表述中能看出来在关系型数据库中的存储同一份数据需要3行，而在Hbase是一行的，而且同一个列族是在同一个Store里面的，更加方便查询。

3、修改pom.xml文件

  1.     <properties>
  2.         <scala.version>2.11</scala.version>
  3.         <spark.version>2.1.1</spark.version>
  4.         <hadoop.version>2.7.7</hadoop.version>
  5.     </properties>
  6.  
  7.     <repositories>
  8.         <repository>
  9.             <id>scala-tools.org</id>
  10.             <name>Scala-Tools Maven2 Repository</name>
  11.             <url>http://scala-tools.org/repo-releases</url>
  12.         </repository>
  13.     </repositories>
  14.  
  15.     <pluginRepositories>
  16.         <pluginRepository>
  17.             <id>scala-tools.org</id>
  18.             <name>Scala-Tools Maven2 Repository</name>
  19.             <url>http://scala-tools.org/repo-releases</url>
  20.         </pluginRepository>
  21.     </pluginRepositories>
  22.  
  23.     <dependencies>
  24.         <dependency>
  25.             <groupId>org.scala-lang</groupId>
  26.             <artifactId>scala-library</artifactId>
  27.             <version>2.11.8</version>
  28. <!--            <scope>provided</scope>-->
  29.         </dependency>
  30.  
  31.         <dependency>
  32.             <groupId>org.apache.spark</groupId>
  33.             <artifactId>spark-core_${scala.version}</artifactId>
  34.             <version>${spark.version}</version>
  35. <!--            <scope>provided</scope>-->
  36.         </dependency>
  37.         <dependency>
  38.             <groupId>org.apache.spark</groupId>
  39.             <artifactId>spark-sql_${scala.version}</artifactId>
  40.             <version>${spark.version}</version>
  41. <!--            <scope>provided</scope>-->
  42.         </dependency>
  43.         <dependency>
  44.             <groupId>log4j</groupId>
  45.             <artifactId>log4j</artifactId>
  46.             <version>1.2.14</version>
  47.         </dependency>
  48.  
  49.         <dependency>
  50.             <groupId>junit</groupId>
  51.             <artifactId>junit</artifactId>
  52.             <version>4.4</version>
  53.             <!--<scope>test</scope>-->
  54.         </dependency>
  55.         <dependency>
  56.             <groupId>org.specs</groupId>
  57.             <artifactId>specs</artifactId>
  58.             <version>1.2.5</version>
  59.             <scope>test</scope>
  60.         </dependency>
  61.         <dependency>
  62.             <groupId>org.scala-lang</groupId>
  63.             <artifactId>scala-reflect</artifactId>
  64.             <version>2.11.8</version>
  65.         </dependency>
  66.         <dependency>
  67.             <groupId>org.scala-lang</groupId>
  68.             <artifactId>scala-compiler</artifactId>
  69.             <version>2.11.8</version>
  70.         </dependency>
  71.         <dependency>
  72.             <groupId>org.scala-lang</groupId>
  73.             <artifactId>scala-library</artifactId>
  74.             <version>2.11.8</version>
  75.         </dependency>
  76.         <dependency>
  77.             <groupId>org.apache.hbase</groupId>
  78.             <artifactId>hbase-client</artifactId>
  79.             <version>1.1.2</version>
  80.         </dependency>
  81.  
  82.         <dependency>
  83.             <groupId>org.apache.hbase</groupId>
  84.             <artifactId>hbase-common</artifactId>
  85.             <version>1.1.2</version>
  86.         </dependency>
  87.  
  88.         <dependency>
  89.             <groupId>org.apache.hbase</groupId>
  90.             <artifactId>hbase-server</artifactId>
  91.             <version>1.1.2</version>
  92.         </dependency>

主要是添加hbase-client、hbase-server、hbase-common，并将spark版本修改为2.1.1。

3、spark读写Hbase测试

往Hbase里面写数据：

  • 通过HTable中put方法：

  1. package Flink
  2. import org.apache.hadoop.conf.Configuration
  3. import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
  4. import org.apache.hadoop.hbase.client.{HBaseAdmin, HTable, Put, Result}
  5. import org.apache.hadoop.hbase.mapred.TableOutputFormat
  6. import org.apache.hadoop.hbase.util.Bytes
  7. import org.apache.log4j.{Level, Logger}
  8. import org.apache.spark.{SparkConf, SparkContext}
  9.  
  10. import org.apache.hadoop.hbase.client.Put
  11. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  12. import org.apache.hadoop.mapred.JobConf
  13. /** *
  14.   *
  15.   * @autor gaowei
  16.   * @Date 2020-07-28 17:55 
  17.   */
  18. object HbaseT1 {
  19.   def getHBaseConfiguration(quorum:String, port:String, tableName:String) = {
  20.     val conf = HBaseConfiguration.create()
  21.     conf.set("hbase.zookeeper.quorum",quorum)
  22.     conf.set("hbase.zookeeper.property.clientPort",port)
  23.  
  24.     conf
  25.   }
  26.   def getHBaseAdmin(conf:Configuration,tableName:String) = {
  27.     val admin = new HBaseAdmin(conf)
  28.     if (!admin.isTableAvailable(tableName)) {
  29.       val tableDesc = new HTableDescriptor(TableName.valueOf(tableName))
  30.       admin.createTable(tableDesc)
  31.     }
  32.  
  33.     admin
  34.   }
  35.   def getTable(conf:Configuration,tableName:String) = {
  36.     new HTable(conf,tableName)
  37.   }
  38.   def main(args: Array[String]) {
  39.     // 屏蔽不必要的日志显示在终端上
  40.     Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  41.  
  42.     val sparkConf = new SparkConf().setAppName("HBaseWriteTest").setMaster("local[2]")
  43.     val sc = new SparkContext(sparkConf)
  44.  
  45.     val tableName = "TEST.USER_INFO"
  46.     val quorum = "localhost"
  47.     val port = "2181"
  48.  
  49.     // 配置相关信息
  50.  
  51.     val conf = getHBaseConfiguration(quorum,port,tableName)
  52.     conf.set(TableOutputFormat.OUTPUT_TABLE,tableName)
  53.  
  54.  
  55.     val indataRDD = sc.makeRDD(Array("R6,jack,16,t1", "R7,Lucy,15,t2", "R8,mike,17,t3", "R9,Lily,14,t4"))
  56.     
  57.     indataRDD.foreachPartition(x=> {
  58.       val conf = getHBaseConfiguration(quorum,port,tableName)
  59.       conf.set(TableOutputFormat.OUTPUT_TABLE,tableName)
  60.  
  61.       val htable = getTable(conf,tableName)
  62.  
  63.       x.foreach(y => {
  64.         val arr = y.split(",")
  65.         val key = arr(0)
  66.         val value = arr(1)
  67.         val value1 = arr(2)
  68.         val value2 = arr(3)
  69.  
  70.         val put = new Put(Bytes.toBytes(key))
  71.         put.add(Bytes.toBytes("INFO"),Bytes.toBytes("C1"),Bytes.toBytes(value))
  72.         put.add(Bytes.toBytes("INFO"),Bytes.toBytes("C2"),Bytes.toBytes(value1))
  73.         put.add(Bytes.toBytes("INFO"),Bytes.toBytes("C3"),Bytes.toBytes(value2))
  74.         htable.put(put)
  75.       })
  76.     })
  77.     sc.stop()
  78.   }
  79.  
  80. }

  • TableOutputFormat向HBase写数据

  1. package Flink
  2. import org.apache.hadoop.conf.Configuration
  3. import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
  4. import org.apache.hadoop.hbase.client.{HBaseAdmin, HTable, Put, Result}
  5. import org.apache.hadoop.hbase.mapred.TableOutputFormat
  6. import org.apache.hadoop.hbase.util.Bytes
  7. import org.apache.log4j.{Level, Logger}
  8. import org.apache.spark.{SparkConf, SparkContext}
  9.  
  10. import org.apache.hadoop.hbase.client.Put
  11. import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  12. import org.apache.hadoop.mapred.JobConf
  13. /** *
  14.   *
  15.   * @autor gaowei
  16.   * @Date 2020-07-28 17:55 
  17.   */
  18. object HbaseT1 {
  19.   def getHBaseConfiguration(quorum:String, port:String, tableName:String) = {
  20.     val conf = HBaseConfiguration.create()
  21.     conf.set("hbase.zookeeper.quorum",quorum)
  22.     conf.set("hbase.zookeeper.property.clientPort",port)
  23.  
  24.     conf
  25.   }
  26.   def getHBaseAdmin(conf:Configuration,tableName:String) = {
  27.     val admin = new HBaseAdmin(conf)
  28.     if (!admin.isTableAvailable(tableName)) {
  29.       val tableDesc = new HTableDescriptor(TableName.valueOf(tableName))
  30.       admin.createTable(tableDesc)
  31.     }
  32.  
  33.     admin
  34.   }
  35.   def getTable(conf:Configuration,tableName:String) = {
  36.     new HTable(conf,tableName)
  37.   }
  38.   def main(args: Array[String]) {
  39.     // 屏蔽不必要的日志显示在终端上
  40.     Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  41.  
  42.     val sparkConf = new SparkConf().setAppName("HBaseWriteTest").setMaster("local[2]")
  43.     val sc = new SparkContext(sparkConf)
  44.  
  45.     val tableName = "TEST.USER_INFO"
  46.     val quorum = "localhost"
  47.     val port = "2181"
  48.  
  49.     // 配置相关信息
  50.  
  51.     val conf = getHBaseConfiguration(quorum,port,tableName)
  52.     conf.set(TableOutputFormat.OUTPUT_TABLE,tableName)
  53.  
  54.  
  55.     val indataRDD = sc.makeRDD(Array("R6,jack,16,t1", "R7,Lucy,15,t2", "R8,mike,17,t3", "R9,Lily,14,t4"))
  56.  
  57.     val jobConf = new JobConf()
  58.     jobConf.setOutputFormat(classOf[TableOutputFormat])
  59.     jobConf.set(TableOutputFormat.OUTPUT_TABLE,tableName)
  60.     indataRDD.map(_.split(",")).map{arr => {
  61.       val put = new Put(Bytes.toBytes(arr(0)))
  62.       put.add(Bytes.toBytes("INFO"),Bytes.toBytes("C1"),Bytes.toBytes(arr(1)))
  63.       put.add(Bytes.toBytes("INFO"),Bytes.toBytes("C2"),Bytes.toBytes(arr(2)))
  64.       put.add(Bytes.toBytes("INFO"),Bytes.toBytes("C3"),Bytes.toBytes(arr(3)))
  65.       (new ImmutableBytesWritable,put)
  66.     }}.saveAsHadoopDataset(jobConf)
  67.     sc.stop()
  68.   }
  69.  
  70. }

4、读Hbase的数据

  1. package Flink
  2. import org.apache.hadoop.hbase.mapreduce.TableInputFormat
  3. import org.apache.hadoop.hbase.util.Bytes
  4. import org.apache.hadoop.conf.Configuration
  5. import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
  6. import org.apache.hadoop.hbase.client.{HBaseAdmin, HTable}
  7. import org.apache.log4j.{Level, Logger}
  8. import org.apache.spark.{SparkConf, SparkContext}
  9. /** *
  10.   *
  11.   * @autor gaowei
  12.   * @Date 2020-07-27 17:06 
  13.   */
  14. object HbseTest {
  15.  
  16.   def getHBaseConfiguration(quorum:String, port:String, tableName:String) = {
  17.     val conf = HBaseConfiguration.create()
  18.     conf.set("hbase.zookeeper.quorum",quorum)
  19.     conf.set("hbase.zookeeper.property.clientPort",port)
  20.  
  21.     conf
  22.   }
  23.   def getHBaseAdmin(conf:Configuration,tableName:String) = {
  24.     val admin = new HBaseAdmin(conf)
  25.     if (!admin.isTableAvailable(tableName)) {
  26.       val tableDesc = new HTableDescriptor(TableName.valueOf(tableName))
  27.       admin.createTable(tableDesc)
  28.     }
  29.  
  30.     admin
  31.   }
  32.   def getTable(conf:Configuration,tableName:String) = {
  33.     new HTable(conf,tableName)
  34.   }
  35.  
  36.   def main(args: Array[String]): Unit = {
  37.     Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  38.     val sparkConf = new SparkConf().setAppName("HBaseReadTest").setMaster("local[2]")
  39.     val sc = new SparkContext(sparkConf)
  40.  
  41.     val tableName = "TEST.USER_INFO"
  42.     val quorum = "localhost"
  43.     val port = "2181"
  44.  
  45.     // 配置相关信息
  46.     val conf = getHBaseConfiguration(quorum,port,tableName)
  47.     conf.set(TableInputFormat.INPUT_TABLE,tableName)
  48.  
  49.     // HBase数据转成RDD
  50.     val hBaseRDD = sc.newAPIHadoopRDD(conf,classOf[TableInputFormat],
  51.       classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
  52.       classOf[org.apache.hadoop.hbase.client.Result]).cache()
  53.  
  54.     // RDD数据操作
  55.     val data = hBaseRDD.map(x => {
  56.       val result = x._2
  57.       val key = Bytes.toString(result.getRow)
  58.       val value = Bytes.toString(result.getValue("INFO".getBytes,"C1".getBytes))
  59.       val value1 = Bytes.toString(result.getValue("INFO".getBytes,"C2".getBytes))
  60.       val value2 = Bytes.toString(result.getValue("INFO".getBytes,"C3".getBytes))
  61.       (key,value,value1,value2)
  62.     })
  63.  
  64.     data.foreach(println)
  65.  
  66.     sc.stop()
  67.   }
  68.  
  69. }

结果：

去Hbase查询的结果：

结论：

Hadoop本质上是：分布式文件系统(HDFS) + 分布式计算框架(Mapreduce) + 调度系统Yarn搭建起来的分布式大数据处理框架。

Hive：是一个基于Hadoop的数据仓库，适用于一些高延迟性的应用（离线开发），可以将结构化的数据文件映射为一张数据库表，并提供简单的sql查询功能。

Hive可以认为是MapReduce的一个包装，把好写的HQL转换为的MapReduce程序，本身不存储和计算数据，它完全依赖于HDFS和MapReduce，Hive中的表是纯逻辑表。hive需要用到hdfs存储文件，需要用到MapReduce计算框架。

HBase：是一个Hadoop的数据库，一个分布式、可扩展、大数据的存储。hbase是物理表，不是逻辑表，提供一个超大的内存hash表，搜索引擎通过它来存储索引，方便查询操作。

HBase可以认为是HDFS的一个包装。他的本质是数据存储，是个NoSql数据库；HBase部署于HDFS之上，并且克服了hdfs在随机读写方面的缺点，提高查询效率。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小惠珠哦/article/detail/874999
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号