大数据Sqoop借助Hive将Mysql数据导入至Hbase

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
云数据库 RDS MySQL,高可用系列 2核4GB
简介: 大数据Sqoop借助Hive将Mysql数据导入至Hbase

1. BulkLoad 介绍

  • 目标
  • 理解 BulkLoad 的大致原理
  • 步骤
  1. 为什么要抽取
  2. 为什么使用 BulkLoad

1.1. 为什么要抽取

大数据Sqoop快速入门

因为数据仓库是甲方自建的, 所以如果我们需要数仓中的数据, 需要申请, 申请完成后, 甲方会将对应的 Hive 表开放给我们, 所以我们需要把 Hive 表中需要的数据抽取到我们的 HBase 中, 如下

抽取方向: Hive -> HBase


1.2. 为什么使用 BulkLoad

在大量数据需要写入HBase时,通常有 put方式和bulkLoad 两种方式。

1、put方式为单条插入,在put数据时会先将数据的更新操作信息和数据信息 写入WAL ,在写入到WAL后, 数据就会被放到MemStore中 ,当MemStore满后数据就会被 flush到磁盘(即形成HFile文件) ,在这种写操作过程会涉及到flush、split、compaction等操作,容易造成节点不稳定,数据导入慢,耗费资源等问题,在海量数据的导入过程极大的消耗了系统性能,避免这些问题最好的方法就是使用BulkLoad的方式来加载数据到HBase中。


b9312adc2de34a8c8ba3652abddc354f.png

2、BulkLoader利用HBase数据按照HFile格式存储在HDFS的原理,使用MapReduce直接批量

生成HFile格式文件后,RegionServers再将HFile文件移动到相应的Region目录下。

1)、Extract,异构数据源数据导入到 HDFS 之上。

2)、Transform,通过用户代码,可以是 MR 或者 Spark 任务将数据转化为 HFile。

3)、Load,HFile 通过 loadIncrementalHFiles 调用将 HFile 放置到 Region 对应的

HDFS 目录上,该过程可能涉及到文件切分。

1、不会触发WAL预写日志,当表还没有数据时进行数据导入不会产生Flush和Split。

2、减少接口调用的消耗,是一种快速写入的优化方式。

Spark读写HBase之使用Spark自带的API以及使用Bulk Load将大量数据导入HBase:

https://www.jianshu.com/p/b6c5a5ba30af

  • 直接使用 HBase 的 Java 客户端, 一条一条插入 HBase 的
  • 使用 BulkLoad

先说说要使用 BulkLoad 的原因

  • 从 Hive 抽取数据到 HBase 是将全量的数据抽取到 HBase, 日增量大概 260 G
  • 如果把这五千万条数据一条一条的插入 HBase, 会影响 HBase 的运行
  • 如果把这五千万条数据一条一条的插入 HBase, 会非常的慢

为什么会非常慢呢? 因为一条数据插入 HBase 的大致步骤如下

  1. 查询元数据表, 得到所要插入的表的 Region 信息
  2. 插入到对应 Region 的 WAL 预写日志
  3. 插入到 Region 中的 Memstore
  1. 达到条件后, Memstore 把数据刷写为 HFile
  2. 达到条件后, 触发 Minor Compaction
  3. 达到条件后, 触发 Major Compaction
  4. 达到条件后, 分裂 Region
  1. 达到条件后, 分裂 Region
  2. 分区再平衡

而我们有 260G 的数据要插入, 触发很多次 Compaction, 会分裂 Region 几百次, 这无疑会造成 HBase 集群的不稳定, 并且, 我们插入的速度也会很慢

所以, 当一次性要插入的数据太多时, 要通过 HBase 的 BulkLoad 方式加载

  1. Spark 读取 Hive 的数据
  1. 生成 HFile
  2. 通过 HBase 的 BulkLoad API 把 HFile 直接交给 RegionServer
  3. RegionServer 切分, 直接放入对应的 Region

1.3 hive导入Hbase

将MySQL数据库中表的数据导入到Hive表中,以便加载到HBase表中。

启动HiveMetastore服务和HiveServer2服务,使用beeline命令行连接,相关命令如下:

[root@bigdata-cdh01 ~]# /export/servers/hive/bin/beeline
Beeline version 1.1.0-cdh5.14.0 by Apache Hive
beeline> !connect jdbc:hive2://bigdata-cdh01.itcast.cn:10000
scan complete in 2ms
Connecting to jdbc:hive2://bigdata-cdh01.itcast.cn:10000
Enter username for jdbc:hive2://bigdata-cdh01.itcast.cn:10000: root
Enter password for jdbc:hive2://bigdata-cdh01.itcast.cn:10000: ****
Connected to: Apache Hive (version 1.1.0-cdh5.14.0)
Driver: Hive JDBC (version 1.1.0-cdh5.14.0)
Transaction isolation: TRANSACTION_REPEATABLE_READ
0: jdbc:hive2://bigdata-cdh01.itcast.cn:10000>

1.3.1 创建表

创建Hive中数据库Database:

CREATE DATABASE tags_dat;

根据MySQL数据库表在Hive数据仓库中构建相应的表:

用户信息表: tbl_users

/export/servers/sqoop/bin/sqoop create-hive-table \
--connect jdbc:mysql://bigdata-cdh01.oldlu.cn:3306/tags_dat \
--table tbl_users \
--username root \
--password 123456 \
--hive-table tags_dat.tbl_users \
--fields-terminated-by '\t' \
--lines-terminated-by '\n'

1.3.2 导入数据至Hive表

使用Sqoop将MySQL数据库表中的数据导入到Hive表中(本质就是存储在HDFS上),具体命

令如下:

用户信息表: tbl_users

/export/servers/sqoop/bin/sqoop import \
--connect jdbc:mysql://bigdata-cdh01.itcast.cn:3306/tags_dat \
--username root \
--password 123456 \
--table tbl_users \
--direct \
--hive-overwrite \
--delete-target-dir \
--fields-terminated-by '\t' \
--lines-terminated-by '\n' \
--hive-table tags_dat.tbl_users \
--hive-import \
--num-mappers 1

2. 从 Hive 中抽取数据到 HBase

  • 目标
  • 将 Hive 的表抽取到 HBase 中
  • 步骤
  1. 准备数据
  1. 导入 MySQL
  2. 导入 Hive
  1. 建立工程 tag-data
  2. 编写 Spark 任务
  3. 运行任务
  4. 查看结果

2.1. 准备数据

2.1.1. 将数据导入到 MySQL

步骤:

  1. 打开 SQL 脚本
  2. 运行 SQL 脚本

MySQL 密码 : itcastmysqlroot

详细解释:

  1. 通过 IDEA 打开脚本, 文件位置在 files/tags_data.sql

  1. 运行脚本

  1. 等待结果

2.1.2. 将数据导入到 Hive

步骤:

  1. 编写 Sqoop 任务脚本
  2. 通过 Hue 上传脚本
  3. 创建 Hive 数据库
  4. 创建 Oozie job 执行脚本, 导入数据

详细解释:

  1. 编写 Sqoop 任务脚本
#!/bin/sh
sqoop import \
  --hive-import \
  --create-hive-table \
  --hive-table tags_data.tbl_goods \
  --connect "jdbc:mysql://master01:3306/tags_dat" \
  --username root \
  --password itcastmysqlroot \
  --query "SELECT * FROM tags_dat.tbl_goods WHERE \$CONDITIONS" \
  --split-by id \
  --direct \
  --target-dir /user/admin/hive/tags_dat \
  --m 2
sqoop import \
  --hive-import \
  --create-hive-table \
  --hive-table tags_data.tbl_goods_new \
  --connect "jdbc:mysql://master01:3306/tags_dat" \
  --username root \
  --password itcastmysqlroot \
  --query "SELECT * FROM tags_dat.tbl_goods_new WHERE \$CONDITIONS" \
  --split-by id \
  --direct \
  --target-dir /user/admin/hive/tags_dat \
  --m 2
sqoop import \
  --hive-import \
  --create-hive-table \
  --hive-table tags_data.tbl_logs \
  --connect "jdbc:mysql://master01:3306/tags_dat?useUnicode=true" \
  --username root \
  --password itcastmysqlroot \
  --query "SELECT * FROM tags_dat.tbl_logs WHERE \$CONDITIONS" \
  --split-by id \
  --direct \
  --target-dir /user/admin/hive/tags_dat \
  --m 2
sqoop import \
  --hive-import \
  --create-hive-table \
  --hive-table tags_data.tbl_orders \
  --connect "jdbc:mysql://master01:3306/tags_dat" \
  --username root \
  --password itcastmysqlroot \
  --query "SELECT * FROM tags_dat.tbl_orders WHERE \$CONDITIONS" \
  --split-by id \
  --direct \
  --target-dir /user/admin/hive/tags_dat \
  --m 2
sqoop import \
  --hive-import \
  --create-hive-table \
  --hive-table tags_data.tbl_users \
  --connect "jdbc:mysql://master01:3306/tags_dat" \
  --username root \
  --password itcastmysqlroot \
  --query "SELECT * FROM tags_dat.tbl_users WHERE \$CONDITIONS" \
  --split-by id \
  --direct \
  --target-dir /user/admin/hive/tags_dat \
  --m 2
  1. 通过 Hue 上传脚本文件

  1. 创建 Hive 数据库

  1. 创建 Oozie Job 执行脚本

  1. 查看执行结果

2.2. 工程配置

数据抽取:

  1. 导入 MySQL
  2. MySQL to Hive
  3. Hive to HBase
  1. 建立工程, 导入 Maven 配置
  2. 代码编写, 通过 Spark 读取 Hive 数据, 落地成 HFile, 通过 BulkLoad 加载到 HBase
  3. 提交, 运行

步骤:

  1. 配置 Maven
  2. 导入 HBase 的配置文件到resource目录中
  • 配置文件在 Files/hbase_conf

从目标来看, 这个工程中需要 Spark, HBase, Hive 的相关依赖, pom.xml 如下

<dependencies>
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-mllib_2.11</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-hive_2.11</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-server</artifactId>
        <exclusions>
            <exclusion>
                <artifactId>jersey-container-servlet-core</artifactId>
                <groupId>org.glassfish.jersey.containers</groupId>
            </exclusion>
            <exclusion>
                <artifactId>guice-servlet</artifactId>
                <groupId>com.google.inject.extensions</groupId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-mapreduce</artifactId>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
    </dependency>
</dependencies>
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
        </plugin>
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
        </plugin>
    </plugins>
</build>

2.3. Spark 任务

  • 步骤
  1. 从 Hive 中读取表数据
  2. 创建 HBase 和 Hadoop 的配置对象
  3. 将数据以 HFile 的形式写入到 HDFS 中
  1. 通过 HBase Client API 将数据 BulkLoad 到 HBase 对应的表中
  • 注意点
  • 如果希望程序要通用一些, 我们可以把 Hive 的表明和 RowKey 列的信息通过 main 方法传进来
  • 要先写 HFile 再 BulkLoad 才是 BulkLoad
  • Job 和 HBaseClient 的配置是模板代码, 不需要记忆
  • add framework support 的时候, 加载 Scala 的 SDK 不能使用 Maven 的 SDK, 要使用自己电脑上安装的 Scala SDK
object HiveToHBase {
  val defaultCF = "default"
  val defaultNameSpace = "default"
  val tempFileDir = "/user/admin/Spark/extra_temp/"
  def main(args: Array[String]): Unit = {
    if (args.length < 3) {
      return
    }
    val sourceDBName = args(0)
    val sourceTableName = args(1)
    val rkeyField = args(2)
    val conf = HBaseConfiguration.create
    conf.set(TableOutputFormat.OUTPUT_TABLE, sourceTableName)
    conf.set("hbase.mapreduce.hfileoutputformat.table.name", sourceTableName)
    val job = Job.getInstance(conf)
    job.setMapOutputKeyClass(classOf[ImmutableBytesWritable])
    job.setMapOutputValueClass(classOf[KeyValue])
    val hfilePath = tempFileDir + sourceTableName
    hive2HFile(sourceDBName, sourceTableName, rkeyField, defaultCF, conf, hfilePath)
    bulkLoad2Table(job, hfilePath, defaultNameSpace, sourceTableName, defaultCF)
  }
  def hive2HFile(sourceDB: String, sourceTable: String, rkeyField: String, cf: String, hadoopConfig: Configuration, hfilePath: String): Unit = {
    val fs = FileSystem.get(hadoopConfig)
    if (fs.exists(new Path(hfilePath))) {
      fs.delete(new Path(hfilePath), true)
    }
    val spark = SparkSession.builder()
      .appName("bulk load from hive")
      .enableHiveSupport()
      .getOrCreate()
    spark.read
      .table(sourceDB + "." + sourceTable)
      .rdd
      .filter(row => row.getAs(rkeyField) != null)
      .flatMap(row => {
        val cfBytes = Bytes.toBytes(cf)
        val rowKeyBytes = Bytes.toBytes(row.getAs(rkeyField).toString)
        row.schema
          .sortBy(field => field.name)
          .map(field => {
            val fieldNameBytes = Bytes.toBytes(field.name)
            val valueBytes = Bytes.toBytes(row.getAs(field.name).toString)
            val kv = new KeyValue(rowKeyBytes, cfBytes, fieldNameBytes, valueBytes)
            (new ImmutableBytesWritable(rowKeyBytes), kv)
          })
      })
      .filter(item => item != null)
      .saveAsNewAPIHadoopFile(
        hfilePath,
        classOf[ImmutableBytesWritable],
        classOf[KeyValue],
        classOf[HFileOutputFormat2],
        hadoopConfig
      )
  }
  def bulkLoad2Table(job: Job, hfilePath: String, namespace: String, name: String, cf: String): Unit = {
    val connection = ConnectionFactory.createConnection(job.getConfiguration)
    val admin = connection.getAdmin
    val tableName = TableName.valueOf(Bytes.toBytes(namespace), Bytes.toBytes(name))
    if (!admin.tableExists(tableName)) {
      admin.createTable(
        TableDescriptorBuilder.newBuilder(tableName)
          .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).build())
          .build()
      )
    }
    val table = connection.getTable(tableName)
    val regionLocator = new HRegionLocator(tableName, connection.asInstanceOf[ClusterConnection])
    HFileOutputFormat2.configureIncrementalLoad(job, table, regionLocator)
    val loader = new LoadIncrementalHFiles(job.getConfiguration)
    loader.doBulkLoad(new Path(hfilePath), admin, table, regionLocator)
  }
}

2.4. 运行任务

一共有五张表需要导入

Hive table name RowKey field
商品表 tbl_goods id
商品表_new tbl_goods_new id
日志表 tbl_logs id
订单表 tbl_orders id
用户表 tbl_users id

需要五个 Oozie Job 去调度执行, 创建方式如下

  1. 打包 Spark 程序
  2. 上传 Spark Jar 包到 HDFS
  3. 创建 Workflow
  • 选择 Spark Jar 包
  • 加入参数, 一个是要导入 HBase 的 Hive 表名, 一个是表中的 RowKey 列
  • 创建 Cordinator
  • Cordinator 会每天执行一次导入增量数据的任务
  • 为了尽快看到效果, 可以先运行 Workflow
  • 如果是导入日增量数据, 可以在 Sqoop 任务的 SQL 中过滤当日数据

详细介绍如下

  1. 打包 Spark 程序


  1. 上传 Spark Jar 包

  1. 创建 Workflow

  1. 创建 Cordinator
相关实践学习
lindorm多模间数据无缝流转
展现了Lindorm多模融合能力——用kafka API写入,无缝流转在各引擎内进行数据存储和计算的实验。
云数据库HBase版使用教程
&nbsp; 相关的阿里云产品:云数据库 HBase 版 面向大数据领域的一站式NoSQL服务,100%兼容开源HBase并深度扩展,支持海量数据下的实时存储、高并发吞吐、轻SQL分析、全文检索、时序时空查询等能力,是风控、推荐、广告、物联网、车联网、Feeds流、数据大屏等场景首选数据库,是为淘宝、支付宝、菜鸟等众多阿里核心业务提供关键支撑的数据库。 了解产品详情:&nbsp;https://cn.aliyun.com/product/hbase &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
相关文章
|
3月前
|
SQL DataWorks 关系型数据库
DataWorks操作报错合集之如何处理数据同步时(mysql->hive)报:Render instance failed
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
|
4月前
|
存储 分布式数据库 数据库
Hbase学习二:Hbase数据特点和架构特点
Hbase学习二:Hbase数据特点和架构特点
67 0
|
26天前
|
SQL 分布式计算 关系型数据库
Hadoop-21 Sqoop 数据迁移工具 简介与环境配置 云服务器 ETL工具 MySQL与Hive数据互相迁移 导入导出
Hadoop-21 Sqoop 数据迁移工具 简介与环境配置 云服务器 ETL工具 MySQL与Hive数据互相迁移 导入导出
47 3
|
26天前
|
SQL 分布式计算 关系型数据库
Hadoop-24 Sqoop迁移 MySQL到Hive 与 Hive到MySQL SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
Hadoop-24 Sqoop迁移 MySQL到Hive 与 Hive到MySQL SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
68 0
|
3月前
|
SQL 存储 关系型数据库
从Hive建表语句到MySQL的转换
【8月更文挑战第11天】
183 7
|
4月前
|
缓存 监控 Shell
使用 HBase Shell 进行数据的实时监控和备份
使用 HBase Shell 进行数据的实时监控和备份
|
4月前
|
Shell 分布式数据库 Hbase
使用 HBase Shell 进行数据的批量导入和导出
使用 HBase Shell 进行数据的批量导入和导出
546 6
|
3月前
|
存储 分布式计算 分布式数据库
《HBase MapReduce之旅:我的学习笔记与心得》——跟随我的步伐,一同探索HBase世界,揭开MapReduce的神秘面纱,分享那些挑战与收获,让你在数据的海洋里畅游无阻!
【8月更文挑战第17天】HBase是Apache顶级项目,作为Bigtable的开源版,它是一个非关系型、分布式数据库,具备高可扩展性和性能。结合HDFS存储和MapReduce计算框架,以及Zookeeper协同服务,HBase支持海量数据高效管理。MapReduce通过将任务拆解并在集群上并行执行,极大提升处理速度。学习HBase MapReduce涉及理解其数据模型、编程模型及应用实践,虽然充满挑战,但收获颇丰,对职业发展大有裨益。
42 0
|
6月前
|
SQL 分布式计算 监控
Sqoop数据迁移工具使用与优化技巧:面试经验与必备知识点解析
【4月更文挑战第9天】本文深入解析Sqoop的使用、优化及面试策略。内容涵盖Sqoop基础,包括安装配置、命令行操作、与Hadoop生态集成和连接器配置。讨论数据迁移优化技巧,如数据切分、压缩编码、转换过滤及性能监控。此外,还涉及面试中对Sqoop与其他ETL工具的对比、实际项目挑战及未来发展趋势的讨论。通过代码示例展示了从MySQL到HDFS的数据迁移。本文旨在帮助读者在面试中展现Sqoop技术实力。
438 2
|
数据采集 SQL 分布式计算
数据处理 、大数据、数据抽取 ETL 工具 DataX 、Kettle、Sqoop
数据处理 、大数据、数据抽取 ETL 工具 DataX 、Kettle、Sqoop
1390 0