【Spark集成HBase】Spark读写HBase表

摘要

本文介绍如何使用 Spark 2.3.2 实现对 HBase 1.4.8 表的读写操作，通过 Scala 语言将 CSV 数据写入 HBase，并利用 Spark SQL 分析数据。代码示例涵盖数据批量写入、全表扫描、数据类型转换及结构化查询，适合大数据开发人员快速掌握 Spark 与 HBase 的集成方法。

一、实验环境准备

1. 技术版本

• Spark：2.3.2
• HBase：1.4.8
• Scala：2.11
• 开发工具：IntelliJ IDEA
• 依赖管理：Maven

2. Maven 依赖配置

在 pom.xml 中添加以下依赖：

<dependencies>  <!-- Spark 核心与 SQL -->  <dependency>  <groupId>org.apache.spark</groupId>  <artifactId>spark-core_2.11</artifactId>  <version>2.3.2</version>  </dependency>  <dependency>  <groupId>org.apache.spark</groupId>  <artifactId>spark-sql_2.11</artifactId>  <version>2.3.2</version>  </dependency>  <!-- HBase 客户端与 MapReduce 支持 -->  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-client</artifactId>  <version>1.4.8</version>  </dependency>  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-common</artifactId>  <version>1.4.8</version>  </dependency>  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-server</artifactId>  <version>1.4.8</version>  </dependency>  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-mapreduce</artifactId>  <version>1.4.8</version>  </dependency>  <!-- Hadoop 客户端（与 HBase 兼容） -->  <dependency>  <groupId>org.apache.hadoop</groupId>  <artifactId>hadoop-client</artifactId>  <version>2.7.3</version>  </dependency>  
</dependencies>  

二、实验步骤

1. 数据准备

• 文件路径：D:\\JavaProjects\\SparkAllProjects\\data\\emp.csv
• 数据格式（示例）：

  7369,SMITH,CLERK,7902,1980-12-17,800,,20  
  7499,ALLEN,SALESMAN,7698,1981-02-20,1600,300,30  
  

  字段含义：员工ID,姓名,职位,上级ID,入职日期,薪资,奖金,部门ID

2. HBase 表结构设计

• 表名：employee
• 列族：info
• 列标识：
  • ename：姓名
  • job：职位
  • mgr：上级ID
  • hiredate：入职日期
  • salary：薪资
  • comm：奖金
  • deptNo：部门ID

3. 代码实现

3.1 数据写入 HBase（writeDataToHBase 方法）

• 核心逻辑：
  1. 读取 CSV 文件为 DataFrame。
  2. 按分区遍历数据，批量创建 Put 对象。
  3. 通过 HBase 连接将数据写入表中，避免单条写入性能瓶颈。
• 核心代码

    def writeDataToHBase(spark: SparkSession): Unit = {// 2.读取数据文件val empDF = spark.read.csv("file:///D:\\JavaProjects\\SparkAllProjects\\data\\emp.csv")// 3.按照DataFrame分区写入HBase表中empDF.foreachPartition(p => {// 3.1 配置HBase连接地址：初始化conf配置对象、配置zk连接地址及其端口val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")conf.set("hbase.zookeeper.property.clientPort", "2181")// 3.2 在每个HBase节点中创建HBase的链接对象val conn = ConnectionFactory.createConnection(conf)// 3.3 获取HBase目标表val table = conn.getTable(TableName.valueOf(TABLE_NAME))// 批量提交时，使用list列表存储put，当达到batchSize大小时提交一次val batchSize = 14 // emp.csv就14条记录，所以设置成14条提交一次var puts = List[Put]() // puts数组// 3.4 将dataframe中的每个分区数据写入table表中try {p.foreach(row => {// 3.5 获取每行中的各个列的数据val empNo = row.getString(0)val ename = row.getString(1)val job = row.getString(2)val mgr = row.getString(3)val hireDate = row.getString(4)val salary = row.getString(5)val comm = row.getString(6)val deptNo = row.getString(7)var mgrStr = "0"if (mgr != null) {mgrStr = mgr}var commStr = "0.0"if (comm != null) {commStr = comm}// 3.6 设置rowkey：按照empNoval rowKey = Bytes.toBytes(empNo)// 3.7 创建Put对象，设置列族中的列和字段val put = new Put(rowKey)put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename"), Bytes.toBytes(ename))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("job"), Bytes.toBytes(job))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr"), Bytes.toBytes(mgrStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate"), Bytes.toBytes(hireDate))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary"), Bytes.toBytes(salary))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm"), Bytes.toBytes(commStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo"), Bytes.toBytes(deptNo))// 3.8 将该Put对象加入Table对象中puts = put :: putsif (puts.size >= batchSize) {// 转成Java中的ArrayListval javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)// table调用put添加javaListstable.put(javaLists)// 添加完成后，清空puts = List[Put]()}// 处理剩余数据: 当不满足14条数据时，把剩余数据写入HBase表中if (puts.nonEmpty) {val javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)table.put(javaLists)}})} finally {// 确保资源释放if (table != null) table.close()if (conn != null) conn.close()}})}
  

3.2 数据读取与分析（readHBaseData 方法

• 核心逻辑：
  1. 使用 ResultScanner 全表扫描 HBase 数据。
  2. 将二进制数据转换为样例类 Employee，自动推断 DataFrame 的 Schema。
  3. 通过 Spark SQL 执行聚合查询（如按部门统计薪资总和）。
• 核心代码：

  case class Employee(  employee_id: Int,  employee_name: String,  job_title: String,  manager_id: Int,  hire_date: String,  salary: Double,  bonus: Double,  department_id: Int  
  )  def readHBaseData(spark: SparkSession): DataFrame = {// 1. 配置 HBase 连接参数val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")  // 替换为你的 ZK 地址conf.set("hbase.zookeeper.property.clientPort", "2181")// 2. 创建 HBase 连接和表对象val conn = ConnectionFactory.createConnection(conf)val table = conn.getTable(TableName.valueOf(TABLE_NAME))var scanner: ResultScanner = nulltry {// 3. 构造扫描器（Scan）并配置val scan = new Scan().addFamily(Bytes.toBytes(CF_NAME))  // 读取指定列族下的所有列.setCaching(500)  // 提升批量读取性能.setCacheBlocks(false)// 4. 获取扫描结果迭代器scanner = table.getScanner(scan)// 5. 遍历结果并转换为 Employee 对象val employees = ListBuffer[Employee]()val it = scanner.iterator()while (it.hasNext) {val result: Result = it.next()// 提取行键（假设 rowkey 是 employee_id 的字符串形式）val rowKeyStr = Bytes.toString(result.getRow)val employeeId = rowKeyStr.toInt  // 转换为 Int（需确保 rowkey 是数字）// 提取各列数据（根据 HBase 实际存储的列名调整）val name = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename")))val job = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("job")))val mgrStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr")))val hireDate = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate")))val salaryStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary")))val commStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm")))val deptNoStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo")))// 处理可能的空值或转换异常（示例：默认值为 0）val managerId = if (mgrStr != null && mgrStr.nonEmpty) mgrStr.toInt else 0val salary = if (salaryStr != null && salaryStr.nonEmpty) salaryStr.toDouble else 0val bonus = if (commStr != null && commStr.nonEmpty) commStr.toDouble else 0val departmentId = if (deptNoStr != null && deptNoStr.nonEmpty) deptNoStr.toInt else 0// 创建 Employee 对象并添加到列表employees += Employee(employee_id = employeeId,employee_name = name,job_title = job,manager_id = managerId,hire_date = hireDate,salary = salary,bonus = bonus,department_id = departmentId)}// 6. 将 Employee 列表转换为 DataFrame（自动推断 Schema）import spark.implicits._spark.createDataFrame(employees.toList)} catch {case e: Exception =>println(s"读取 HBase 数据失败: ${e.getMessage}")throw e  // 抛异常终止流程} finally {// 7. 释放所有资源（关键！避免连接泄漏）if (scanner != null) scanner.close()if (table != null) table.close()if (conn != null) conn.close()}}
  

3.3 Spark SQL 分析

• 读取数据后，通过 Spark SQL 执行聚合查询：

  val df = readHBaseData(spark)  
  df.createOrReplaceTempView("emp")  
  spark.sql("""  SELECT department_id,  SUM(salary + bonus) AS total  FROM emp  GROUP BY department_id  ORDER BY total DESC  
  """).show(false)  
  

• 其他分析查询请自行操作。

3.4 完整代码

• 在IDEA中创建名为WriteAndReadDataToHBase单例对象
• 添加如下完成代码：

  package com.lpssfxy.spark.datasourceimport org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
  import org.apache.hadoop.hbase.client.{ConnectionFactory, Put, Result, ResultScanner, Scan}
  import org.apache.hadoop.hbase.util.Bytes
  import org.apache.spark.sql.{DataFrame, SparkSession}import java.util
  import scala.collection.mutable.ListBuffercase class Employee(employee_id: Int,       // 对应 HBase rowkey（假设 rowkey 是员工ID）employee_name: String,  // 对应列族 info:enamejob_title: String,      // 对应列族 info:jobmanager_id: Int,        // 对应列族 info:mgrhire_date: String,      // 对应列族 info:hiredatesalary: Double,            // 对应列族 info:salarybonus: Double,             // 对应列族 info:commdepartment_id: Int      // 对应列族 info:deptNo)/*** spark读写HBase表employee*/
  object WriteAndReadDataToHBase {// 定义表名称及其列族名称private val TABLE_NAME = "employee"private val CF_NAME = "info"def main(args: Array[String]): Unit = {// 1. 准备环境：SparkSession初始化val spark = SparkSession.builder().appName("WriteDataToHBase").master("local[*]").getOrCreate()// 2.调用writeDataToHBase方法将文件写入HBase中//writeDataToHBase(spark)// 3. 读HBase数据val df = readHBaseData(spark)//df.show(false)df.createOrReplaceTempView("emp")spark.sql("select department_id,sum(salary+bonus) as total from emp group by department_id order by total desc").show(false)// 4.停止SparkSession对象，释放资源spark.stop()}/*** 写数据到HBase表中* @param spark*/def writeDataToHBase(spark: SparkSession): Unit = {// 2.读取数据文件val empDF = spark.read.csv("file:///D:\\JavaProjects\\SparkAllProjects\\data\\emp.csv")// 3.按照DataFrame分区写入HBase表中empDF.foreachPartition(p => {// 3.1 配置HBase连接地址：初始化conf配置对象、配置zk连接地址及其端口val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")conf.set("hbase.zookeeper.property.clientPort", "2181")// 3.2 在每个HBase节点中创建HBase的链接对象val conn = ConnectionFactory.createConnection(conf)// 3.3 获取HBase目标表val table = conn.getTable(TableName.valueOf(TABLE_NAME))// 批量提交时，使用list列表存储put，当达到batchSize大小时提交一次val batchSize = 14 // emp.csv就14条记录，所以设置成14条提交一次var puts = List[Put]() // puts数组// 3.4 将dataframe中的每个分区数据写入table表中try {p.foreach(row => {// 3.5 获取每行中的各个列的数据val empNo = row.getString(0)val ename = row.getString(1)val job = row.getString(2)val mgr = row.getString(3)val hireDate = row.getString(4)val salary = row.getString(5)val comm = row.getString(6)val deptNo = row.getString(7)var mgrStr = "0"if (mgr != null) {mgrStr = mgr}var commStr = "0.0"if (comm != null) {commStr = comm}// 3.6 设置rowkey：按照empNoval rowKey = Bytes.toBytes(empNo)// 3.7 创建Put对象，设置列族中的列和字段val put = new Put(rowKey)put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename"), Bytes.toBytes(ename))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("job"), Bytes.toBytes(job))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr"), Bytes.toBytes(mgrStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate"), Bytes.toBytes(hireDate))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary"), Bytes.toBytes(salary))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm"), Bytes.toBytes(commStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo"), Bytes.toBytes(deptNo))// 3.8 将该Put对象加入Table对象中puts = put :: putsif (puts.size >= batchSize) {// 转成Java中的ArrayListval javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)// table调用put添加javaListstable.put(javaLists)// 添加完成后，清空puts = List[Put]()}// 处理剩余数据: 当不满足14条数据时，把剩余数据写入HBase表中if (puts.nonEmpty) {val javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)table.put(javaLists)}})} finally {// 确保资源释放if (table != null) table.close()if (conn != null) conn.close()}})}/*** 读取HBase表数据* @param spark* @return*/def readHBaseData(spark: SparkSession): DataFrame = {// 1. 配置 HBase 连接参数val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")  // 替换为你的 ZK 地址conf.set("hbase.zookeeper.property.clientPort", "2181")// 2. 创建 HBase 连接和表对象val conn = ConnectionFactory.createConnection(conf)val table = conn.getTable(TableName.valueOf(TABLE_NAME))var scanner: ResultScanner = nulltry {// 3. 构造扫描器（Scan）并配置val scan = new Scan().addFamily(Bytes.toBytes(CF_NAME))  // 读取指定列族下的所有列.setCaching(500)  // 提升批量读取性能.setCacheBlocks(false)// 4. 获取扫描结果迭代器scanner = table.getScanner(scan)// 5. 遍历结果并转换为 Employee 对象val employees = ListBuffer[Employee]()val it = scanner.iterator()while (it.hasNext) {val result: Result = it.next()// 提取行键（假设 rowkey 是 employee_id 的字符串形式）val rowKeyStr = Bytes.toString(result.getRow)val employeeId = rowKeyStr.toInt  // 转换为 Int（需确保 rowkey 是数字）// 提取各列数据（根据 HBase 实际存储的列名调整）val name = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename")))val job = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("job")))val mgrStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr")))val hireDate = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate")))val salaryStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary")))val commStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm")))val deptNoStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo")))// 处理可能的空值或转换异常（示例：默认值为 0）val managerId = if (mgrStr != null && mgrStr.nonEmpty) mgrStr.toInt else 0val salary = if (salaryStr != null && salaryStr.nonEmpty) salaryStr.toDouble else 0val bonus = if (commStr != null && commStr.nonEmpty) commStr.toDouble else 0val departmentId = if (deptNoStr != null && deptNoStr.nonEmpty) deptNoStr.toInt else 0// 创建 Employee 对象并添加到列表employees += Employee(employee_id = employeeId,employee_name = name,job_title = job,manager_id = managerId,hire_date = hireDate,salary = salary,bonus = bonus,department_id = departmentId)}// 6. 将 Employee 列表转换为 DataFrame（自动推断 Schema）import spark.implicits._spark.createDataFrame(employees.toList)} catch {case e: Exception =>println(s"读取 HBase 数据失败: ${e.getMessage}")throw e  // 抛异常终止流程} finally {// 7. 释放所有资源（关键！避免连接泄漏）if (scanner != null) scanner.close()if (table != null) table.close()if (conn != null) conn.close()}}}

三、实验结果

1. 数据写入验证

• 控制台输出无异常日志，HBase 表 employee 中生成对应 rowkey 的记录。
• 通过 HBase Shell 命令 scan 'employee' 可查看数据：

  hbase> scan 'employee', {LIMIT => 2}  
  ROW       COLUMN+CELL                                              
  7369      column=info:ename, timestamp=... value=SMITH               
  7369      column=info:job, timestamp=... value=CLERK               
  # ...（其他列省略）  
  

2. 数据读取与分析结果

• 原始数据展示：

  +-----------+-------------+-----------+-----------+----------+------+-----+-------------+  
  |employee_id|employee_name|job_title  |manager_id |hire_date |salary|bonus|department_id|  
  +-----------+-------------+-----------+-----------+----------+------+-----+-------------+  
  |7369       |SMITH        |CLERK      |7902       |1980-12-17|800.0 |0.0  |20           |  
  |7499       |ALLEN        |SALESMAN   |7698       |1981-02-20|1600.0|300.0|30           |  
  # ...（其他行省略）  
  

• Spark SQL 聚合结果：

  +-------------+------------+  
  |department_id|total       |  
  +-------------+------------+  
  |30           |20150.0     |  
  |20           |10875.0     |  
  |10           |8750.0      |  
  +-------------+------------+  
  

四、常见问题与优化

1. 空值处理

• HBase 列值为 null 时，result.getValue 返回 null，需通过 Option 或判空逻辑处理：

  val commStr = Option(result.getValue(...)).map(Bytes.toString).getOrElse("0.0")  
  

2. 性能优化

• 批量写入：调整 batchSize（建议 500-2000），减少 RPC 调用次数。
• 扫描缓存：通过 scan.setCaching(500) 提升全表扫描效率。
• 数据类型：避免过度使用字符串类型，对数值字段直接存储二进制数据（如 Bytes.toDouble）。

3. 集群配置（可选做）

• 在 spark-submit 中添加 HBase 配置文件（如 hbase-site.xml），确保 Executor 节点访问 ZooKeeper：

  spark-submit --files hbase-site.xml ...  
  

五、总结

本文通过实际案例演示了 Spark 与 HBase 的集成流程，实现了从 CSV 数据写入 HBase 到结构化数据分析的完整链路。核心要点包括：

1. 使用 foreachPartition 实现批量写入，避免单条操作性能损耗。
2. 通过 ResultScanner 和样例类映射，简化 HBase 数据到 DataFrame 的转换。
3. 利用 Spark SQL 对 HBase 数据进行高效分析，发挥分布式计算优势。

该实验适用于海量结构化数据的存储与分析场景，可进一步扩展至实时数据处理或机器学习模型训练。