Spark与HBase的集成

比眉伴天荒 2022-12-27 12:36 194阅读 0赞

### 目录 ###

*   *  一、Spark与HBase的集成
     *  二、Spark SQL与HBase的集成

**版本说明：**

*  hbase版本：hbase-1.3.1
 *  spark版本：spark-2.4.7-bin-hadoop2.7

## 一、Spark与HBase的集成 ##

**背景：**  
Spark支持多种数据源，但是Spark对HBase的读写都没有相对优雅的api，但spark和HBase整合的场景又比较多，故通过spark的数据源API自己实现了一套比较方便操作HBase的API。

**数据模型：**

row,addres,age,username
    
    001,guangzhou,20,alex
    002,shenzhen,34,jack
    003,beijing,23,lili

**需求分析：**  
通过spark读取hbase中的数据或者将数据写入到hbase中。

**添加配置：**  
在pom.xml文件中添加如下配置：

<dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-client</artifactId>
                <version>${hbase.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-common</artifactId>
                <version>${hbase.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-server</artifactId>
                <version>${hbase.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-log4j12</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-common</artifactId>
                <version>${hadoop.version}</version>
            </dependency>

**代码实现：**

package com.kfk.spark.sql
    
    import com.kfk.spark.common.CommSparkSessionScala
    import org.apache.hadoop.conf.Configuration
    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.mapred.TableOutputFormat
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.hadoop.mapred.JobConf
    import org.apache.spark.sql.SparkSession
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/10 * @time : 8:53 下午 */
    object HBaseSpark { 
        def main(args: Array[String]): Unit = { 
    
            // configuration
            val spark = CommSparkSessionScala.getSparkSession()
            val hbaseConf = HBaseConfiguration.create()
            hbaseConf.set("hbase.zookeeper.property.clientPort","2181")
            hbaseConf.set("hbase.zookeeper.quorum","bigdata-pro-m04")
            hbaseConf.set("hbase.master","bigdata-pro-m04:60010")
    
            // get
            getHBase(hbaseConf,spark)
    
            // write
            writeHBase(hbaseConf,spark)
    
        }
    
        /** * 读取hbase中的数据 * @param hbaseConf * @param spark */
        def getHBase(hbaseConf : Configuration,spark : SparkSession): Unit ={ 
    
            // 获取表名
            hbaseConf.set(TableInputFormat.INPUT_TABLE,"stu")
    
            // 将hbase中的数据转换成rdd
            val hbaseRDD =  spark.sparkContext.newAPIHadoopRDD(hbaseConf,
                classOf[TableInputFormat],
                classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
                classOf[org.apache.hadoop.hbase.client.Result])
    
            // 打印数据
            hbaseRDD.foreach(result => { 
                val key = Bytes.toString(result._2.getRow)
                val addres = Bytes.toString(result._2.getValue("info".getBytes(),"addres".getBytes()))
                val age = Bytes.toString(result._2.getValue("info".getBytes(),"age".getBytes()))
                val username = Bytes.toString(result._2.getValue("info".getBytes(),"username".getBytes()))
    
                println("row key:" + key + " addres=" + addres + " age=" + age + " username=" + username)hashCode()
    
                /** * row key:001 addres=guangzhou age=20 username=alex * row key:002 addres=shenzhen age=34 username=jack * row key:003 addres=beijing age=23 username=lili */
    
            })
        }
    
        /** * 将数据写入到hbase * @param hbaseConf * @param spark */
        def writeHBase(hbaseConf : Configuration,spark : SparkSession): Unit ={ 
    
            // 初始化job，设置输出格式，TableOutputFormat 是 org.apache.hadoop.hbase.mapred 包下的
            val jobConf = new JobConf(hbaseConf)
    
            jobConf.setOutputFormat(classOf[TableOutputFormat])
    
            // 获取表名
            jobConf.set(TableOutputFormat.OUTPUT_TABLE,"stu")
    
            // 准备数据
            val array = Array("004,shanghai,25,jone",
                              "005,nanjing,31,cherry",
                              "006,wuhan,18,pony")
    
            val rdd = spark.sparkContext.makeRDD(array)
    
            // 将写入到hbase的数据转换成rdd
            val saveRDD = rdd.map(line => line.split(",")).map(x => { 
    
                /** * 一个Put对象就是一行记录，在构造方法中指定主键 * 所有插入的数据 须用 org.apache.hadoop.hbase.util.Bytes.toBytes 转换 * Put.addColumn 方法接收三个参数：列族，列名，数据 */
                val put = new Put(Bytes.toBytes(x(0)))
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("addres"),Bytes.toBytes(x(1)))
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"),Bytes.toBytes(x(2)))
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("username"),Bytes.toBytes(x(3)))
                (new ImmutableBytesWritable,put)
            })
    
            // 写入到hbase中
            saveRDD.saveAsHadoopDataset(jobConf)
    
        }
    }

运行结果：

row key:001 addres=guangzhou age=20 username=alex
    row key:002 addres=shenzhen age=34 username=jack
    row key:003 addres=beijing age=23 username=lili
    row key:004 addres=shanghai age=25 username=jone
    row key:005 addres=nanjing age=31 username=cherry
    row key:006 addres=wuhan age=18 username=pony

本来源数据只有前三行，通过写入后三行，再打印出结果。

## 二、Spark SQL与HBase的集成 ##

**Spark SQL与HBase集成，其核心就是Spark Sql通过hive外部表来获取HBase的表数据。**

将hbase、hive、mysql相关jar包拷贝到spark的jars目录下

hbase:
    hbase-client-1.3.1.jar
    hbase-common-1.3.1.jar
    hbase-protocol-1.3.1.jar
    hbase-server-1.3.1.jar
    metrics-core-2.2.0.jar
    
    hive:
    hive-hbase-handler-2.3.3.jar
    htrace-core-3.1.0-incubating.jar
    
    mysql:
    mysql-connector-java-5.1.48-bin.jar

创建与HBase集成的Hive的外部表：

CREATE EXTERNAL TABLE stu(
    id string,
    addres string,
    age string,
    username string) 
    STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
    WITH SERDEPROPERTIES (
    "hbase.columns.mapping" = 
    ":key,info:addres,info:age,info:username") 
    TBLPROPERTIES ("hbase.table.name" = "stu");

关于hive和hbase的集成请看这篇文章：  
[https://blog.csdn.net/weixin\_45366499/article/details/110955459][https_blog.csdn.net_weixin_45366499_article_details_110955459]

**启动hive中的metastore：**

bin/hive --service metastore

**启动spark-shell：**

val df = spark.sql("select * from hivespark.stu").show

+---+---------+---+--------+
    | id|   addres|age|username|
    +---+---------+---+--------+
    |001|guangzhou| 20|    alex|
    |002| shenzhen| 34|    jack|
    |003|  beijing| 23|    lili|
    |004| shanghai| 25|    jone|
    |005|  nanjing| 31|  cherry|
    |006|    wuhan| 18|    pony|
    +---+---------+---+--------+

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[https_blog.csdn.net_weixin_45366499_article_details_110955459]: https://blog.csdn.net/weixin_45366499/article/details/110955459

发表评论取消回复

表情：

评论列表（有 0 条评论，194人围观）

还没有评论，来说两句吧...

相关阅读

相关 HBase与MapReduce集成

HBase集成MapReduce 通过 HBase 的相关 JavaAPI，我们可以实现伴随 HBase 操作的 MapReduce 过程，比如使用MapReduce 将

￡神魔★判官ぃ/ 2023年10月04日 16:44/ 0 赞/ 4 阅读

相关 Spark SQL与JDBC的集成

数据源： userid,addres,age,username 001,guangzhou,20,alex 002,shenzhen,34,

向右看齐/ 2022年12月27日 14:02/ 0 赞/ 117 阅读

相关 Spark与HBase的集成

目录一、Spark与HBase的集成二、Spark SQL与HBase的集成版本说明： hbase版本：hbase-1.3.1

比眉伴天荒/ 2022年12月27日 12:36/ 0 赞/ 195 阅读

相关 Hive与HBase的集成

Hive提供了与HBase的集成，使得能够在HBase表上使用HQL语句进行查询插入操作以及进行Join和Union等复杂查询、同时也可以将hive表中的数据映射到Hbase

柔情只为你懂/ 2022年12月27日 11:14/ 0 赞/ 213 阅读

相关 HBase与MR的集成

MR <--- 读取 HBase的数据 MR ---> 写入 HBase（使用场景：数据迁移） HBase

r囧r小猫/ 2022年12月20日 12:48/ 0 赞/ 161 阅读

相关 spark与hive集成的问题

ambari默认安装的spark，是无法访问hive的。需要把 /etc/hive/conf 中的 hive-site.xml 移到spark的conf目录下并修改以下

Dear 丶/ 2022年12月03日 09:25/ 0 赞/ 175 阅读

相关 Flink SQL与HBase的集成

版本说明： flink-1.12.1 hbase-1.4.13 目录（1）Flink SQL与HBase的集成配置（2）测试

太过爱你忘了你带给我的痛/ 2022年11月18日 05:39/ 0 赞/ 238 阅读

相关 hive与hbase集成

`` `简介` 这篇文章最初是基于介绍[HIVE-705][]。这个功能允许Hive QL命令访问HBase表，进行读（select）、写（insert）操作。它甚至可

我就是我/ 2022年08月26日 08:12/ 0 赞/ 205 阅读

相关 hive与hbase集成

[hive与hbase集成][hive_hbase] 这篇文章最初是基于介绍[HIVE-705][]。这个功能允许Hive QL命令访问HBase表，进行读（select）、

迈不过友情╰/ 2022年08月07日 12:56/ 0 赞/ 226 阅读

相关 HBase 与 MapReduce 集成

6. HBase 与 MapReduce 集成 6.1 官方 HBase 与 MapReduce 集成 1. 查看 HBase 的 MapReduce 任务的执行

朴灿烈づ我的快乐病毒、/ 2022年01月06日 15:07/ 0 赞/ 282 阅读