Spark On Hbase的官方jar包編譯與使用

前言

找了一番，Spark讀寫HBase已經有專門的 Maven 依賴包可用,HBase提供了一個HBase Spark Connector項目，hbase官網文檔提到這個項目可從源碼編譯。這樣就有相似spark-kafka,spark-hive的hbase-spark依賴了。mvn庫如今提供了一個1.0版本https://mvnrepository.com/artifact/org.apache.hbase.connectors.spark/hbase-spark/1.0.0，其Spark爲2.4.0，Scala爲2.11.12，其餘版本須要自行編譯。

官網1.0jar maven依賴：

<!-- https://mvnrepository.com/artifact/org.apache.hbase.connectors.spark/hbase-spark -->
<dependency>
    <groupId>org.apache.hbase.connectors.spark</groupId>
    <artifactId>hbase-spark</artifactId>
    <version>1.0.0</version>
</dependency>

複製代碼

很是容易和這個項目混淆：https://mvnrepository.com/artifact/org.apache.hbase/hbase-spark，這個不是HBase Connector Spark項目！

編譯 hbase-spark 源碼

源碼地址

Apache Hbase維護的項目，今後處下載源碼壓縮包： Hbase Connectors-Spark源碼

Apache HBase™ Spark Connector

• Scala and Spark Versions

To generate an artifact for a different spark version and/or scala version, pass command-line options as follows (changing version numbers appropriately):

$ mvn -Dspark.version=2.3.1 -Dscala.version=2.11.8 -Dscala.binary.version=2.11 clean install
複製代碼

準備編譯出Spark2.3.1&Scala2.11.8的hbase-spark依賴:

unzip hbase-connectors-master.zip
cd hbase-connectors-master/
mvn -Dspark.version=2.3.1 -Dscala.version=2.11.8 -Dscala.binary.version=2.11 clean install
複製代碼

編譯報錯

1. maven版本太低，安裝一下maven3.5.4，配置環境變量便可。
2. TestJavaHBaseContext Failed,編譯時添加 -DskipTests便可

   

   最終編譯語句：

mvn -Dspark.version=2.3.1 -Dscala.version=2.11.8 -Dscala.binary.version=2.11 -DskipTests clean install

編譯成功

位置：~/hbase-connectors-master/spark/hbase-spark/target

就是 hbase-spark-1.0.1-SNAPSHOT.jar，如今能夠在項目中使用了。

使用hbase-spark-1.0.1-SNAPSHOT.jar

使用HBaseContext寫數據

import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}

//從編譯的hbase-spark-1.0.1-SNAPSHOT.jar中引入
import org.apache.hadoop.hbase.spark.HBaseContext

import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import java.util.UUID
/** * 引入了從Hbase官網編譯的hbase-spark jar,調用HBaseContext * Spark批量寫數據到HBase */
object SparkWithHBase {
  def main(args: Array[String]): Unit = {
    //Spark統一入口
    val spark = SparkSession.builder()
      .appName("Spark JDBC Test")
      .master("local")
      .getOrCreate()
    //列族名稱
    val SRC_FAMILYCOLUMN = "info"
    //Hbase配置
    val config = HBaseConfiguration.create()
    config.set("hbase.zookeeper.quorum", "manager.bigdata")
    config.set("hbase.zookeeper.property.clientPort", "2181")
    //Hbase上下文，是API的核心
    val hbaseContext = new HBaseContext(spark.sparkContext, config)
    //讀取數據源，封裝成<RowKey,Values>這種格式
    val rdd: RDD[(String, Array[(String, String)])] = spark.read.csv("hdfs://manager.bigdata:8020/traffic.txt")
      .rdd
      .map(r => {
        (UUID.randomUUID().toString, Array((r.getString(0), "c1"), (r.getString(1), "c2"), (r.getString(2), "c3")))
      })
    //使用批量put方法寫入數據
    hbaseContext.bulkPut[(String, Array[(String, String)])](rdd,
      TableName.valueOf("spark_hbase_bulk_put"),
      row => {
        val put = new Put(Bytes.toBytes(row._1))
        row._2.foreach(putValue => put.addColumn(
          Bytes.toBytes(SRC_FAMILYCOLUMN),
          Bytes.toBytes(putValue._2),
          Bytes.toBytes(putValue._1)))
        put
      })

  }
}
複製代碼

經查詢，數據成功寫入HBase。