Spark - RDD使用详解10（HBase的读取与写入）

十一、HBase 的读取与写入

1，准备工作

<!-- hbase-client 依赖-->
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>2.5.5</version>
</dependency>
<!-- HBase MapReduce 模块的依赖-->
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-mapreduce</artifactId>
    <version>2.5.5</version>
</dependency>

import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put, Get, Admin,
  TableDescriptorBuilder, ColumnFamilyDescriptorBuilder}
import org.apache.hadoop.hbase.util.Bytes

object Hello {
  def main(args: Array[String]): Unit = {
    // 配置 HBase 连接信息
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.60.9")
    conf.set("hbase.zookeeper.property.clientPort", "2181")

    // 建立 HBase 连接
    val connection: Connection = ConnectionFactory.createConnection(conf)
    val admin: Admin = connection.getAdmin
    val tableName = TableName.valueOf("my_table")
    val table = connection.getTable(tableName)

    try {
      // 自动创建表（如果不存在）
      if (!admin.tableExists(tableName)) {
        val tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
          .addColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("cf")).build())
          .build()
        admin.createTable(tableDescriptor)
        println("成功创建表 my_table。")
      }

      // 写入数据到 HBase
      val put = new Put(Bytes.toBytes("row1"))
      put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col1"), Bytes.toBytes("hangge.com"))
      table.put(put)
      println("成功写入数据到 HBase。")

      // 从 HBase 读取数据
      val get = new Get(Bytes.toBytes("row1"))
      val result = table.get(get)
      val value = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("col1")))
      println(s"从 HBase 读取数据: $value")
    } finally {
      // 关闭连接
      table.close()
      connection.close()
    }
  }
}

2，将 RDD 的数据写入到 HBase 表中

import org.apache.hadoop.hbase.{HBaseConfiguration}
import org.apache.hadoop.hbase.client.{Put}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.rdd.RDD
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{SparkConf, SparkContext}

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
    // 创建 Spark 上下文环境对象（连接对象）
    val sc: SparkContext = new SparkContext(sparkConf)

    // 配置 HBase 连接信息
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.60.9")
    conf.set("hbase.zookeeper.property.clientPort", "2181")

    // 生成作业
    val jobConf = new JobConf(conf, this.getClass)
    jobConf.set(TableOutputFormat.OUTPUT_TABLE,  "my_table")
    val job = Job.getInstance(jobConf)
    job.setOutputKeyClass(classOf[ImmutableBytesWritable])
    job.setOutputValueClass(classOf[Put])
    job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])

    // 构建记录
    val dataRDD: RDD[String] = sc.makeRDD(Array("1,hangge,M,100", "2,lili,M,27", "3,liuyun,F,35"))

    // 数据进行转换
    val rdd: RDD[(ImmutableBytesWritable, Put)] = dataRDD.map(_.split(',')).map {
      arr: Array[String] => {
        //行健的值
        val put = new Put(Bytes.toBytes(arr(0)))
        //name列的值
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("name"), Bytes.toBytes(arr(1)))
        //gender列的值
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("gender"), Bytes.toBytes(arr(2)))
        //age列的值
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("age"), Bytes.toBytes(arr(3).toInt))
        (new ImmutableBytesWritable, put)
      }
    }

    // 使用 saveAsNewAPIHadoopDataset 方法将RDD数据写入到 HBase 表
    rdd.saveAsNewAPIHadoopDataset(job.getConfiguration)
    println("成功将RDD数据写入到 HBase。")

    //关闭 Spark
    sc.stop()
  }
}

3，读取 HBase 表中数据转换为RDD

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.{Result}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.{TableInputFormat}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.{SparkConf, SparkContext}

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
      .set("spark.hadoopRDD.ignoreEmptySplits", "false")

    // 创建 Spark 上下文环境对象（连接对象）
    val sc: SparkContext = new SparkContext(sparkConf)

    // 配置 HBase 连接信息
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.60.9")
    conf.set("hbase.zookeeper.property.clientPort", "2181")
    conf.set(TableInputFormat.INPUT_TABLE, "my_table")

    // 读取 HBase 数据并将其转换为 RDD
    val hbaseRDD = sc.newAPIHadoopRDD(
      conf,
      classOf[TableInputFormat],
      classOf[ImmutableBytesWritable],
      classOf[Result]
    )

    // 对 RDD 进行转换和统计
    val genderCountRDD = hbaseRDD.map {
      case (_, result) =>
        val gender = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("gender")))
        (gender, 1)
    }.reduceByKey(_ + _)

    // 打印结果
    genderCountRDD.collect().foreach {
      case (gender, count) => println(s"性别: $gender, 数量: $count")
    }

    //关闭 Spark
    sc.stop()
  }
}