SparkSQL 与 Hbase 整合

一、环境

  1. 1) 创建 hive on hbase 数据表, 具体参见 hive-on-hbase 文档
  3. 1) 把 hbase 所有的主机名配置到 /etc/hosts 中,保证每个 host 都能访问到 hbase 集群的服务器
  4.     vim /etc/hosts
  5.     例如 :
  6.     10.10.33.175    uhadoop-ociicy-core1
  7.     10.10.7.68      uhadoop-ociicy-core2
  8.     10.10.43.97     uhadoop-ociicy-core3
  9.     10.10.240.22    uhadoop-ociicy-core4
  10.     10.10.236.241   uhadoop-ociicy-core5
  11.     10.10.222.21    uhadoop-ociicy-core6
  12.     10.10.229.183   uhadoop-ociicy-task3
  13.     10.10.234.131   uhadoop-ociicy-task4
  15. 2) 配置相关的依赖 jar 包(具体跟随集群环境走)
  16.     $HBASE_HOME/lib/hbase-annotations-1.2.0-cdh5.9.2.jar
  17.     $HBASE_HOME/lib/hbase-spark-1.2.0-cdh5.9.2.jar
  18.     $HBASE_HOME/lib/hbase-common-1.2.0-cdh5.9.2.jar
  19.     $HBASE_HOME/lib/hbase-client-1.2.0-cdh5.9.2.jar
  20.     $HBASE_HOME/lib/hbase-server-1.2.0-cdh5.9.2.jar
  21.     $HBASE_HOME/lib/hbase-protocol-1.2.0-cdh5.9.2.jar
  22.     $HBASE_HOME/lib/guava-12.0.1.jar
  23.     $HBASE_HOME/lib/htrace-core-3.2.0-incubating.jar
  24.     $HBASE_HOME/lib/zookeeper.jar
  25.     $HBASE_HOME/lib/protobuf-java-2.5.0.jar
  26.     $HBASE_HOME/lib/hbase-hadoop2-compat-1.2.0-cdh5.9.2.jar
  27.     $HBASE_HOME/lib/hbase-hadoop-compat-1.2.0-cdh5.9.2.jar
  28.     $HBASE_HOME/lib/metrics-core-2.2.0.jar
  29.     $HIVE_HOME/lib/hive-hbase-handler-1.1.0-cdh5.9.0.jar

二、命令

  1. 1. 注意事项
  3. 配置好 hbase-site.xml, 配置好 --jars
  4. spark-sql \
  5. --master yarn \
  6. --deploy-mode client \
  7. --name spark-hbase-demo \
  8. --driver-cores 1 \
  9. --driver-memory 1024M \
  10. --executor-cores 1 \
  11. --executor-memory 1024M \
  12. --num-executors 1 \
  13. --files $HBASE_HOME/conf/hbase-site.xml \
  14. --jars file://$HBASE_HOME/lib/hbase-annotations-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/hbase-spark-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/hbase-common-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/hbase-client-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/hbase-server-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/hbase-protocol-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/guava-12.0.1.jar,file://$HBASE_HOME/lib/htrace-core-3.2.0-incubating.jar,file://$HBASE_HOME/lib/zookeeper.jar,file://$HBASE_HOME/lib/protobuf-java-2.5.0.jar,file://$HBASE_HOME/lib/hbase-hadoop2-compat-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/hbase-hadoop-compat-1.2.0-cdh5.9.2.jar,file://$HBASE_HOME/lib/metrics-core-2.2.0.jar,file://$HIVE_HOME/lib/hive-hbase-handler-1.1.0-cdh5.9.0.jar

三、RDD 编程

使用 Spark On HBase 机制, 在不迁移数据的情况下，使用 Spark 嵌入 HBase 中分析查询，目前有如下几个方案:

  1. hortonworks/shc - spark-on-hbase 方案 实现了 Spark Datasource API ， 直接使用 Spark Catalyst 引擎(spark 2.0 引入)进行查询优化，耦合低，由于使用 API 访问，后续 Spark 版本升级 Spark Catalyst 引擎被优化，性能也会跟着提升。

  2. Huawei-Spark/Spark-SQL-on-HBase - 华为 2015 入侵方案 在 Spark Catalyst 引擎内嵌入自己的查询优化计划, 将 RDD 发送到 HBase，入侵到 HBase 的 Coprocessors 协同处理器中执行任务，例如 Group By。由于此查询计划是自己实现功能的复杂性，不使用 Spark Catalyst 官方优化引擎，所以日后升级、补丁，不跟随 Spark 官方走，会导致日后维护难和不稳定

  3. nerdammer/spark-hbase-connector - nerdammer 对传统读写 Hbase TableInputFormat 和 TableOutputFormat 的封装

  4. cloudera-labs/SparkOnHBase - coluder 2015 方案 (使用 Spark 1.6 版本) cloudera 提供的方案 2015 年方案

  1. import org.apache.spark.sql.{SparkSession, DataFrame, SQLContext}
  2. import org.apache.spark.{SparkConf, SparkContext}
  3. import org.apache.spark.sql.execution.datasources.hbase._
  5. /**
  6.  * SparkRDD 读取 HBase 数据
  7.  * 下载 git@github.com:hortonworks-spark/shc.git
  8.  *    git clone  git@github.com:hortonworks-spark/shc.git
  9.  * 选择 spark 和 hbase 对应版本, 切换到 v1.1.0-2.0 分支,  Hbase 1.1.0, spark 2.0, https://github.com/hortonworks-spark/shc/tree/v1.1.0-2.0
  10.  *    git fetch origin v1.1.0-2.0:v1.1.0-2.0
  11.  *    git checkout v1.1.0-2.0
  12.  * 打包编译, 修改 pom.xml 文件, 修改 spark 版本号信息为 2.0.2
  13.  *    mvn clean package
  14.  * 放到项目 lib 目录下即可
  15.  */
  16. object TestSparkHBase {
  18.      def main(args: Array[String]): Unit = {
  19.         // 初始化上下文
  20.         val sparkConf = new SparkConf()
  21.            //.setMaster("local")
  22.            .setAppName("HBaseExample")
  23.         val sc = new SparkContext(sparkConf)
  24.         val sqlContext = new SQLContext(sc)
  26.         // 隐式转换
  27.         import sqlContext.implicits._
  29.         // 装在 HBase 数据到 DataFrame
  30.         val df = withCatalog(sqlContext, catalog)
  32.         // 读取 RDD
  33.         df.take(10).foreach { x => println(x) }
  35.         // 读取hive
  36.         df.registerTempTable("user_profile")
  37.         sqlContext.sql("SELECT col0,col1,col2,col3 from user_profile LIMIT 10").show
  40.     }
  42.      // 定义 HBase 信息
  43.     def catalog = s"""{
  44.        |"table":{"namespace":"default", "name":"user_profile"},
  45.        |"rowkey":"key",
  46.        |"columns":{
  47.          |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
  48.          |"col1":{"cf":"browser", "col":"channel_id", "type":"string"},
  49.          |"col2":{"cf":"common", "col":"update_date", "type":"string"},
  50.          |"col3":{"cf":"pcsafe", "col":"gc_nav_click", "type":"string"}
  51.        |}
  52.      |}""".stripMargin
  54.      /**
  55.       * 装在 HBbase 数据
  56.       */
  57.      def withCatalog(sqlContext: SQLContext, cat: String): DataFrame = {
  58.         sqlContext
  59.             .read
  60.             .options(Map(HBaseTableCatalog.tableCatalog->cat))
  61.             .format("org.apache.spark.sql.execution.datasources.hbase")
  62.             .load()
  63.      }
  65. }