专栏首页Hadoop实操使用Spark通过BulkLoad快速导入数据到HBase

使用Spark通过BulkLoad快速导入数据到HBase

温馨提示:如果使用电脑查看图片不清晰,可以使用手机打开文章单击文中的图片放大查看高清原图。

Fayson的github: https://github.com/fayson/cdhproject

提示:代码块部分可以左右滑动查看噢

1.文档编写目的


在项目中有需求需要将Hive表中的数据存储在HBase中。使用Spark访问Hive表,将读表数据导入到HBase中,写入HBase有两种方式:一种是通过HBase的API接口批量的将数据写入HBase,另一种是通过BulkLoad的方式生成HFile文件然后加载到HBase中,两种方式相比之下第二种效率会更高。本篇文章Fayson主要介绍如何使用Spark读取Hive表数据通过BulkLoad的方式快速的将数据导入到HBase。

  • 文章概述

1.环境准备

2.示例代码及运行

3.总结

  • 测试环境

1.CM5.14.3和CDH5.14.2

2.集群未启用Sentry和Kerberos

3.Spark1.6.0

2.环境准备


本篇文章主要使用HBase中hbase-spark包提供的HBaseContext来实现,需要准备hbase-spark的依赖包并部署到Spark集群。

1.将准备好的hbase-spark-1.2.0-cdh5.13.1.jar部署到集群所有节点的/opt/cloudera/parcels/CDH/lib/spark/lib目录下

[root@cdh01 ~]# ll /opt/cloudera/parcels/CDH/lib/spark/lib/

(可左右滑动)

2.登录CM配置在spark-env.sh中增加hbase-spark的依赖,增加如下配置内容

export SPARK_DIST_CLASSPATH=$SPARK_DIST_CLASSPATH:/opt/cloudera/parcels/CDH/lib/spark/lib/hbase-spark-1.2.0-cdh5.13.1.jar

(可左右滑动)

保存配置,并重新部署Spark客户端配置

3.Hive表示例数据查看

去除重复数据共600条数据

3.Spark示例代码


1.使用Maven创建Scala示例工程,Pom.xml文件内容如下:

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-assembly_2.10</artifactId>
    <version>1.6.0-cdh5.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-spark</artifactId>
    <version>1.2.0-cdh5.13.1</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.10</artifactId>
    <version>1.6.0-cdh5.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.10</artifactId>
    <version>1.6.0-cdh5.12.1</version>
</dependency>

(可左右滑动)

2.在工程中创建Hive2HBase.scala文件,内容如下:

package com.cloudera.hbase

import java.io.IOException

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.hbase._
import org.apache.hadoop.hbase.client._
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.hbase.mapreduce.{HFileOutputFormat2, LoadIncrementalHFiles}
import org.apache.hadoop.hbase.spark.{HBaseContext, KeyFamilyQualifier}
import org.apache.hadoop.hbase.spark.HBaseRDDFunctions._
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapreduce.Job

import scala.collection.mutable

/**
  * package: com.cloudera.hbase
  * describe: 使用BulkLoad的方式将Hive数据导入HBase
  * creat_user: Fayson 
  * email: htechinfo@163.com
  * creat_date: 2018/7/31
  * creat_time: 下午2:04
  * 公众号:Hadoop实操
  */
object Hive2HBase {

  def main(args: Array[String]) {

    //库名、表名、rowKey对应的字段名、批次时间、需要删除表的时间参数
    val rowKeyField = "id"
    val quorum = "cdh01.fayson.com,cdh02.fayson.com,cdh03.fayson.com"
    val clientPort = "2181"
    val hBaseTempTable = "ods_user_hbase"

    val sparkConf = new SparkConf().setAppName("Hive2HBase")
    val sc = new SparkContext(sparkConf)

    val hiveContext = new HiveContext(sc)
    //从hive表读取数据
    val datahiveDF = hiveContext.sql(s"select * from ods_user")

    //表结构字段
    var fields = datahiveDF.columns

    //去掉rowKey字段
    fields = fields.dropWhile(_ == rowKeyField)

    val hBaseConf = HBaseConfiguration.create()
    hBaseConf.set("hbase.zookeeper.quorum", quorum)
    hBaseConf.set("hbase.zookeeper.property.clientPort", clientPort)

    //表不存在则建Hbase临时表
    creteHTable(hBaseTempTable, hBaseConf)

    val hbaseContext = new HBaseContext(sc, hBaseConf)

    //将DataFrame转换bulkload需要的RDD格式
    val rddnew = datahiveDF.rdd.map(row => {
      val rowKey = row.getAs[String](rowKeyField)

      fields.map(field => {
        val fieldValue = row.getAs[String](field)
        (Bytes.toBytes(rowKey), Array((Bytes.toBytes("info"), Bytes.toBytes(field), Bytes.toBytes(fieldValue))))
      })
    }).flatMap(array => {
      (array)
    })

    //使用HBaseContext的bulkload生成HFile文件
    hbaseContext.bulkLoad[Put](rddnew.map(record => {
      val put = new Put(record._1)
      record._2.foreach((putValue) => put.addColumn(putValue._1, putValue._2, putValue._3))
      put
    }), TableName.valueOf(hBaseTempTable), (t : Put) => putForLoad(t), "/tmp/bulkload")

    val conn = ConnectionFactory.createConnection(hBaseConf)
    val hbTableName = TableName.valueOf(hBaseTempTable.getBytes())
    val regionLocator = new HRegionLocator(hbTableName, classOf[ClusterConnection].cast(conn))
    val realTable = conn.getTable(hbTableName)
    HFileOutputFormat2.configureIncrementalLoad(Job.getInstance(), realTable, regionLocator)

    // bulk load start
    val loader = new LoadIncrementalHFiles(hBaseConf)
    val admin = conn.getAdmin()
    loader.doBulkLoad(new Path("/tmp/bulkload"),admin,realTable,regionLocator)

    sc.stop()
  }

  /**
    * 创建HBase表
    * @param tableName 表名
    */
  def creteHTable(tableName: String, hBaseConf : Configuration) = {
    val connection = ConnectionFactory.createConnection(hBaseConf)
    val hBaseTableName = TableName.valueOf(tableName)
    val admin = connection.getAdmin
    if (!admin.tableExists(hBaseTableName)) {
      val tableDesc = new HTableDescriptor(hBaseTableName)
      tableDesc.addFamily(new HColumnDescriptor("info".getBytes))
      admin.createTable(tableDesc)
    }
    connection.close()
  }

  /**
    * Prepare the Put object for bulkload function.
    * @param put The put object.
    * @throws java.io.IOException
    * @throws java.lang.InterruptedException
    * @return Tuple of (KeyFamilyQualifier, bytes of cell value)*/
  @throws(classOf[IOException])
  @throws(classOf[InterruptedException])
  def putForLoad(put: Put): Iterator[(KeyFamilyQualifier, Array[Byte])] = {
    val ret: mutable.MutableList[(KeyFamilyQualifier, Array[Byte])] = mutable.MutableList()
    import scala.collection.JavaConversions._
    for (cells <- put.getFamilyCellMap.entrySet().iterator()) {
      val family = cells.getKey
      for (value <- cells.getValue) {
        val kfq = new KeyFamilyQualifier(CellUtil.cloneRow(value), family, CellUtil.cloneQualifier(value))
        ret.+=((kfq, CellUtil.cloneValue(value)))
      }
    }
    ret.iterator
  }
}

(可左右滑动)

3.使用Maven命令将工程编译为jar包

4.示例运行


1.将编译好的spark-demo-1.0-SNAPSHOT.jar包上传至服务器,使用spark-submit提交

export HADOOP_USER_NAME=hbase
spark-submit --class com.cloudera.hbase.Hive2HBase \
--master yarn-client \
--driver-cores 1 \
--driver-memory 2g \
--executor-cores 1 \
--executor-memory 2g \
spark-demo-1.0-SNAPSHOT.jar

(可左右滑动)

通过Spark作业界面,查看作业运行情况

2.作业执行成功后,查看HBase表数据

查看表数据

5.总结


1.本篇文章是使用hbase-spark包中提供的bulkload方法生成HFile文件,然后将生成的文件导入到HBase表中。

2.使用bulkload的方式导入数据到HBase表时,在load HFile文件到表过程中会有短暂的时间导致该表停止服务(在load文件过程中需要先disable表,load完成后在enable表。

3.需要使用hbase用户提交Spark作业

GitHub地址:

https://github.com/fayson/cdhproject/blob/master/sparkdemo/src/main/scala/com/cloudera/hbase/Hive2HBase.scala

提示:代码块部分可以左右滑动查看噢

为天地立心,为生民立命,为往圣继绝学,为万世开太平。 温馨提示:如果使用电脑查看图片不清晰,可以使用手机打开文章单击文中的图片放大查看高清原图。

原创文章,欢迎转载,转载请注明:转载自微信公众号Hadoop实操

本文分享自微信公众号 - Hadoop实操(gh_c4c535955d0f)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-08-01

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • PySpark数据类型转换异常分析

    在使用PySpark的SparkSQL读取HDFS的文本文件创建DataFrame时,在做数据类型转换时会出现一些异常,如下:

    Fayson
  • 0753-6.3.3-如何在Redhat7.6安装CDH6.3.3

    CDH6.3.3于2020年2月4日发布,从该版本开始,Cloudera将不再提供CDH免费版,用户只能通过试用或者购买正式许可证才能够下载CDH安装包及使用,...

    Fayson
  • 0539-5.15.0-HBase-Spark无法在Spark2编译通过问题解决

    在Spark2代码中使用hbase-spark依赖包访问HBase时,编写的代码无法完成编译,在编译的过程中提示如下错误:

    Fayson
  • 如何做Spark 版本兼容

    同理对应的Vectors object 也是。这就造成了一个比较大的困难,比如下面的代码就很难做到兼容了,切换Spark就无法通过编译:

    用户2936994
  • Log4j1、Logback 以及 Log4j2 性能测试对

    jdk:1.7.0_79 cpu:i5-4570@3.20GHz 4核 eclipse:3.7 操作系统:win7

    芋道源码
  • 提交Spark任务的三种方式

    在使用Spark的过程中,一般都会经历调试,提交任务等等环节,如果每个环节都可以确认程序的输入结果,那么无疑对加快代码的调试起了很大的作用,现在,借助IDEA可...

    ZONGLYN
  • k8s对接云实现自动扩容pod

    在之前的文章我介绍了下 Custom Metric 怎么实现自动扩容的。k8s基于自定义指标实现自动扩容

    SY小站
  • 小白学Flask第一天 | 我的第一个Flask程序

    相信大部分人进来读这篇文章是因为以前没接触过Flask框架,这里我给大家介绍一下这个框架。

    Python进击者
  • Spark读取CSV异常 java.lang.ArrayIndexOutOfBoundsException:62

    情况1: 将GBK编码的文件转文UTF-8(我碰见的),当然这种情况也可以用情况2中的解决办法解决~

    董可伦
  • Python Flask 优秀资源大全

    版权: https://github.com/humiaozuzu/awesome-flask Awesome Flask ? =============...

    若与

扫码关注云+社区

领取腾讯云代金券