spark 对hbase 操作-Spark-Mac私塾

背景

随着大数据技术的不断发展，越来越多的企业和组织开始利用大数据来指导和加强决策。而在这些大数据技术中，Apache Hadoop 和 Apache Spark 两个开源框架都在数据处理领域占据了重要地位。但是，Hadoop 是一种分布式文件系统，而 Spark 则是一种快速而通用的数据处理引擎。这意味着它们在不同的方面都有自己的优缺点。

在实际的项目实践过程中，我们经常会遇到需要将 Hadoop 中的数据导入到 Spark 中进行处理的场景。然而，由于 Hadoop 是一种文件系统，而 Spark 则是一种数据处理引擎，因此二者之间的数据交互过程并不是很方便。为了实现这一目标，我们需要使用一种中间组件来协调它们之间的数据传输。

本篇文章就将探讨如何使用 Spark 来操作 Hbase，以期提高数据处理效率和精度。

HBase 简介

Hbase 是一个开源的分布式、非关系型数据库，可以支持可扩展的海量数据存储。它主要基于 Google 的 Bigtable 思想，用于存储和处理海量结构化数据，具有高性能、高可靠性等特点。

HBase 中的数据可以按照不同的版本进行存储，而且支持写入时的自动合并，这就使得 HBase 很适用于存储弹性结构的数据。尤其是在需要高速访问海量数据、支持增量扩容、数据版本管理等需求下，HBase 显得尤为优秀。

Spark 简介

Spark 是一款基于内存计算的分布式数据处理引擎。它可以在集群上高效地进行大规模数据处理，并支持多种不同的数据源和数据处理方式。

Spark 提供了多种不同的 API，其中最常用的是 Spark SQL 和 Spark Streaming。Spark SQL 可以用于处理结构化数据，而 Spark Streaming 则可以用于处理流式数据。此外，Spark 还提供了 GraphX 和 MLlib 等组件，可以用于处理图形数据和机器学习任务。

HBase 和 Spark 的结合

要将 HBase 中的数据导入到 Spark 中进行处理，我们需要引入两个库：HBase 和 Spark。在 Spark 中，我们可以使用 Spark HBase Connector，这是一个可以通过 Spark SQL 和 Spark Streaming 直接操作 HBase 数据库的库。它支持读写 HBase 数据库、自定义过滤器，同时还可以自动将 RDD 中的数据存储到 Hbase 表格中。

HBase 与 Spark 的结合可以在以下几方面发挥作用：

利用 Spark 处理 HBase 中的数据，使得数据处理更加高效且应用更加广泛；
利用 HBase 存储 Spark 处理结果，再利用这些结果开发出下一个 Spark 应用程序；
Hbase 提供了支持高扩展性以及更优的可读性的方式处理数据，HBase 成为了分析海量数据的不二之选。

Spark 对 HBase 的操作

HBase 数据导入 Spark

可以使用 Spark 自带的 HadoopInputFormat 和 HadoopOutputFormat 将 HBase 表中的数据导入到 Spark RDD 中，如下：

import org.apache.spark.SparkContext
import org.apache.hadoop.hbase.client.{HBaseAdmin,HTable,Put,Result}
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.mapred.JobConf

val hconf = HBaseConfiguration.create(sc.hadoopConfiguration)
hconf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
val jobConf = new JobConf(hconf, this.getClass)
jobConf.set("mapreduce.output.fileoutputformat.outputdir", outputDir)
jobConf.setOutputFormat(classOf[TableOutputFormat])
jobConf.set(TableOutputFormat.OUTPUT_TABLE, tableName)

val rawData = sc.newAPIHadoopRDD(jobConf,classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

通过上述代码，我们便可以将 HBase 表中的数据导入到 Spark RDD 中；接下来，我们可以使用 Spark 的 API 进行数据处理和分析。可以使用“foreach”遍历所有的行

rawData.foreach(r => {
   val rowKey = Bytes.toString(r._1.get())
   val name = Bytes.toString(r._2.getValue(cf.getBytes(),"name".getBytes()))
   val genders = Bytes.toString(r._2.getValue(cf.getBytes(),"gender".getBytes()))
   //处理数据
})

其中，r._1.get() 和 r._2.get() 分别代表 HBase 表中的行键和行数据。注意，在处理行数据时，需要使用“getValue”方法，并且必须指定数据的“column family”和“qualifier”信息。

Spark 数据写入 HBase

向 HBase 写入数据可以使用 HadoopOutputFormat 和saveAsNewAPIHadoopDataset等函数，如下：


val rdd = sc.parallelize(Seq(
  (2,"test2",30),
  (3,"test3",40),
  (4,"test4",50)
))
val conf = HBaseConfiguration.create()
val jobConf = new JobConf(conf, this.getClass)
jobConf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
val newAPIJobConf = Job.getInstance(jobConf)
newAPIJobConf.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, tableName)

val hbasePuts = rdd.map { case (id, name, age) =>
  val put = new Put(Bytes.toBytes(id))
  put.addColumn(cf.getBytes, "name".getBytes, Bytes.toBytes(name))
  put.addColumn(cf.getBytes, "age

spark 对hbase 操作