spark sql通过hbase-connectors操作hbase示例

我会带着你远行 2023-06-20 06:42 158阅读 0赞

整体思路，通过spark sql加载hbase数据源，借助spark引擎码sql进行查询统计需要的数据。

文档参考：  
[hbase官网spark hbase结合文档][hbase_spark hbase]  
[spark sql示例][spark sql]

## 一. 集成步骤 ##

### 1.1 spark hbase-connectors驱动编译 ###

spark操作hbase需要按照自己环境的版本要求进行编译驱动包。具体参看连接地址。  
[驱动包github连接地址][github]

[已编译的驱动包下载地址][Link 1]

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.2.2 -Dscala.version=2.11.7 -Dscala.binary.version=2.11 clean install

### 1.2 spark hbase pom依赖 ###

本人hbase线上版本为 1.x 版本，但是操作需要使用2.x的hbase驱动包。低版本会报某些错误。  
驱动包本地安装参考 [安装jar到本地库][jar]

<spark.version>2.4.4</spark.version>
        <scala.binary.version>2.11</scala.binary.version>
        <hbase.version>2.2.0</hbase.version>
    
      <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_${scala.binary.version}</artifactId>
          <version>${spark.version}</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-sql_${scala.binary.version}</artifactId>
          <version>${spark.version}</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming_${scala.binary.version}</artifactId>
          <version>${spark.version}</version>
        </dependency>
     <dependency>  
          <groupId>fakepath</groupId>
          <artifactId>hbase-spark</artifactId>
          <version>1.0.1</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.hbase</groupId>
          <artifactId>hbase-client</artifactId>
          <version>${hbase.version}</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.hbase</groupId>
          <artifactId>hbase-common</artifactId>
          <version>${hbase.version}</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.hbase</groupId>
          <artifactId>hbase-server</artifactId>
          <version>${hbase.version}</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.hbase</groupId>
          <artifactId>hbase-mapreduce</artifactId>
          <version>${hbase.version}</version>
        </dependency>

### 1.3 代码配置 ###

JavaSparkContext jsc = xxxxx; //spark context 对象具体创建参看spark官网
     Configuration hbaseConfig = HBaseConfiguration.create();
     hbaseConfig.set("hbase.zookeeper.quorum", "192.168.1.3");
    
     JavaHBaseContext javaHBaseContext = new JavaHBaseContext(jsc, hbaseConfig);
    
     String catalog = "{" +
                    "       \"table\":{\"namespace\":\"default\", \"name\":\"testTable\"},"+
                    "       \"rowkey\":\"key\"," +
                    "       \"columns\":{" +
                    "         \"rowkey\":{\"cf\":\"rowkey\", \"col\":\"key\", \"type\":\"string\"}," +
                    "         \"col\":{\"cf\":\"data\", \"col\":\"col\", \"type\":\"string\"}" +
                    "       }" +
                    "     }";
    // colunmns hbase字段映射配置  cf列族名称 col字段名称  namespace表空间 
    
    
     Dataset<Row> dataSet = sparkSession.read()
                    .format("org.apache.hadoop.hbase.spark")
                    .option(HBaseTableCatalog.tableCatalog(), catalog);
                    .load();
     dataSet.createOrReplaceTempView("dataSet");
     Dataset<Row> sqlRow = sqlContext.sql("select * from dataSet");  //通过sql语句查询hbase数据

[hbase_spark hbase]: https://hbase.apache.org/book.html#spark
[spark sql]: http://spark.apache.org/docs/latest/sql-programming-guide.html
[github]: https://github.com/apache/hbase-connectors/tree/master/spark
[Link 1]: https://download.csdn.net/download/weixin_40294332/12518012
[jar]: https://blog.csdn.net/weixin_40294332/article/details/103453829