【Spark分析HBase数据】Spark读取并分析HBase数据

一、择要

Apache Spark 是一个快速、通用的大数据处置惩罚引擎，提供了丰富的 API 用于数据处置惩罚和分析。HBase 是一个分布式、可扩展的 NoSQL 数据库，适合存储海量结构化和半结构化数据。Spark 与 HBase 的结合可以充分发挥两者的上风，实现高效的数据处置惩罚和分析。
Spark 可以通过 HBase 的 Java API 大概专用的连接器来读取 HBase 中的数据。在读取数据时，Spark 可以将 HBase 表中的数据转换为 RDD（弹性分布式数据集）大概 DataFrame，然后使用 Spark 的各种操作进行数据处置惩罚和分析。
本文以Spark2.3.2读取HBase1.4.8中的hbase_emp_table表数据进行简单分析，用户实现相关的业务逻辑。
二、实现过程

• 在IDEA创建工程SparkReadHBaseData
  
• 在pom.xml文件中添加依赖
  1. <properties>
  2.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  3.     <maven.compiler.source>1.8</maven.compiler.source>
  4.     <maven.compiler.target>1.8</maven.compiler.target>
  5.     <scala.version>2.11.8</scala.version>
  6.     <spark.version>2.3.3</spark.version>
  7.     <hbase.version>1.4.8</hbase.version>
  8. </properties>
  10. <dependencies>
  11.     <!-- Spark 依赖 -->
  12.     <dependency>
  13.         <groupId>org.apache.spark</groupId>
  14.         <artifactId>spark-core_2.11</artifactId>
  15.         <version>${spark.version}</version>
  16.     </dependency>
  18.     <dependency>
  19.         <groupId>org.apache.spark</groupId>
  20.         <artifactId>spark-sql_2.11</artifactId>
  21.         <version>${spark.version}</version>
  22.     </dependency>
  24.     <!-- HBase 依赖 -->
  25.     <dependency>
  26.         <groupId>org.apache.hbase</groupId>
  27.         <artifactId>hbase-client</artifactId>
  28.         <version>${hbase.version}</version>
  29.     </dependency>
  30.     <dependency>
  31.         <groupId>org.apache.hbase</groupId>
  32.         <artifactId>hbase-common</artifactId>
  33.         <version>${hbase.version}</version>
  34.     </dependency>
  35.     <dependency>
  36.         <groupId>org.apache.hbase</groupId>
  37.         <artifactId>hbase-server</artifactId>
  38.         <version>${hbase.version}</version>
  39.     </dependency>
  40.     <dependency>
  41.         <groupId>org.apache.hbase</groupId>
  42.         <artifactId>hbase-hadoop-compat</artifactId>
  43.         <version>${hbase.version}</version>
  44.     </dependency>
  46.     <!-- Hadoop 依赖 -->
  47.     <dependency>
  48.         <groupId>org.apache.hadoop</groupId>
  49.         <artifactId>hadoop-client</artifactId>
  50.         <version>2.7.4</version>
  51.         <scope>provided</scope>
  52.     </dependency>
  54.     <!-- 处理依赖冲突 -->
  55.     <dependency>
  56.         <groupId>com.google.guava</groupId>
  57.         <artifactId>guava</artifactId>
  58.         <version>12.0.1</version>
  59.     </dependency>
  61.     <!-- 使用scala2.11.8进行编译和打包 -->
  62.     <dependency>
  63.         <groupId>org.scala-lang</groupId>
  64.         <artifactId>scala-library</artifactId>
  65.         <version>${scala.version}</version>
  66.     </dependency>
  68. </dependencies>
  70. <build>
  71.     <!-- 指定scala源代码所在的目录 -->
  72.     <sourceDirectory>src/main/scala</sourceDirectory>
  74.     <plugins>
  76.         <!-- Scala 编译插件 -->
  77.         <plugin>
  78.             <groupId>net.alchim31.maven</groupId>
  79.             <artifactId>scala-maven-plugin</artifactId>
  80.             <version>3.4.6</version>
  81.             <executions>
  82.                 <execution>
  83.                     <goals>
  84.                         <goal>compile</goal>
  85.                         <goal>testCompile</goal>
  86.                     </goals>
  87.                 </execution>
  88.             </executions>
  89.         </plugin>
  91.         <plugin>
  92.             <groupId>org.apache.maven.plugins</groupId>
  93.             <artifactId>maven-assembly-plugin</artifactId>
  94.             <version>3.6.0</version>
  95.             <configuration>
  96.                 <archive>
  97.                     <!-- 项目中有多个主类时，采用不指定主类规避pom中只能配置一个主类的问题 -->
  98.                     <manifest/>
  99.                 </archive>
  100.                 <descriptorRefs>
  101.                     <descriptorRef>jar-with-dependencies</descriptorRef>
  102.                 </descriptorRefs>
  103.             </configuration>
  104.             <executions>
  105.                 <execution>
  106.                     <id>make-assembly</id>
  107.                     <phase>package</phase>
  108.                     <goals>
  109.                         <goal>single</goal>
  110.                     </goals>
  111.                 </execution>
  112.             </executions>
  113.         </plugin>
  114.     </plugins>
  115. </build>

  复制代码
• 新建com.lpssfxy的package
  
  
  
  
• 在该package下新建名为SparkReadHBaseData的Object，编写步伐实现业务逻辑：
  1. import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
  2. import org.apache.hadoop.hbase.client.{ConnectionFactory, Scan}
  3. import org.apache.hadoop.hbase.util.Bytes
  4. import org.apache.spark.sql.{SparkSession}
  6. /**
  7. * Employee样例类
  8. *
  9. * @param empNo
  10. * @param eName
  11. * @param job
  12. * @param mgr
  13. * @param hireDate
  14. * @param salary
  15. * @param comm
  16. * @param deptNo
  17. */
  18. case class Employee(empNo: Int, eName: String, job: String, mgr: Int, hireDate: String, salary: Double, comm: Double, deptNo: Int)
  21. object SparkReadHBaseData {
  23.   private val TABLE_NAME = "hbase_emp_table"
  24.   private val INFO_CF = "info"
  26.   def main(args: Array[String]): Unit = {
  27.     val spark = SparkSession.builder()
  28.       .appName("SparkHBaseIntegration")
  29.       .master("local[*]")
  30.       .getOrCreate()
  32.     val conf = HBaseConfiguration.create()
  33.     conf.set("hbase.zookeeper.quorum", "s1,s2,s3")
  34.     conf.set("hbase.zookeeper.property.clientPort", "2181")
  36.     val connection = ConnectionFactory.createConnection(conf)
  37.     val table = connection.getTable(TableName.valueOf(TABLE_NAME))
  39.     val scan = new Scan()
  40.     scan.addFamily(Bytes.toBytes(INFO_CF))
  42.     // 扫描 HBase 表并转换为 RDD
  43.     val results = table.getScanner(scan)
  44.     val data = Iterator.continually(results.next()).takeWhile(_ != null).map { result =>
  45.       val rowKey = Bytes.toString(result.getRow())
  46.       val eName = Bytes.toString(result.getValue(Bytes.toBytes(INFO_CF), Bytes.toBytes("ename")))
  47.       val job = Bytes.toString(result.getValue(Bytes.toBytes(INFO_CF), Bytes.toBytes("job")))
  48.       val mgrString = Bytes.toString(result.getValue(Bytes.toBytes(INFO_CF), Bytes.toBytes("mgr")))
  49.       var mgr: Int = 0
  50.       if (!"".equals(mgrString) && null != mgrString) {
  51.         mgr = mgrString.toInt
  52.       }
  53.       val hireDate = Bytes.toString(result.getValue(Bytes.toBytes(INFO_CF), Bytes.toBytes("hiredate")))
  54.       val salary = Bytes.toString(result.getValue(Bytes.toBytes(INFO_CF), Bytes.toBytes("sal")))
  55.       val commString = Bytes.toString(result.getValue(Bytes.toBytes(INFO_CF), Bytes.toBytes("comm")))
  56.       var comm: Double = 0
  57.       if (!"".equals(commString) && null != commString) {
  58.         comm = commString.toDouble
  59.       }
  60.       val deptNo = Bytes.toString(result.getValue(Bytes.toBytes(INFO_CF), Bytes.toBytes("deptno")))
  61.       (rowKey.toInt, eName, job, mgr, hireDate, salary.toDouble, comm, deptNo.toInt)
  62.     }.toList
  64.     // 转换为 DataFrame
  65.     import spark.implicits._
  66.     val df = spark.sparkContext.parallelize(data).map(item => {
  67.       Employee(item._1, item._2, item._3, item._4, item._5, item._6, item._7, item._8)
  68.     }).toDF()
  70.     // 将df注册成临时表
  71.     df.createOrReplaceTempView("emp")
  72.     // 需求1：统计各个部门总支出
  73.     val totalExpense = spark.sql("select deptNo,sum(salary) as total from emp group by deptNo order by total desc")
  74.     totalExpense.show()
  75.     // 需求2： 统计各个部门总的支出（包括工资和奖金），并按照总支出升序排
  76.         val totalExpense2 = spark.sql("select deptNo,sum(salary + comm) as total from emp group by deptNo order by total")
  77.         totalExpense2.show()
  78.         // TODO：需求3-结合dept部门表来实现多表关联查询，请同学自行实现
  80.     // 关闭连接
  81.     connection.close()
  82.     // 停止spark，释放资源
  83.     spark.stop()
  84.   }
  85. }

  复制代码
• 为了没有大量无关日记输出，在resources目录下新建log4j.properties，添加如下内容：
  1. log4j.rootLogger=ERROR,stdout
  2. # write to stdout
  3. log4j.appender.stdout=org.apache.log4j.ConsoleAppender
  4. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
  5. log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %5p --- [%50t] %-80c(line:%5L) : %m%n

  复制代码
• 启动虚拟机中的hdfs、zookeeper和hbase
  1. start-dfs.sh
  2. zkServer.sh start
  3. start-hbase.sh

  复制代码
• 运行代码，查看实行结果
  
  
  
  

三、小结

• 本实验仅仅演示Spark读取HBase表数据并简单分析的过程，可以作为复杂的业务逻辑分析的基础。
  
• Spark 读取并分析 HBase 数据具有高性能、丰富的数据分析功能、可扩展性、灵活性和实时性等上风。然而，也存在数据一致性、复杂的配置和管理、资源消耗和兼容性等不敷。在实际应用中，必要根据详细的需求和场景来选择是否使用 Spark 和 HBase 的组合，并注意解决大概出现的问题。
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。