Hadoop高手之路10-Sqoop数据迁移

​ 在实际开发中，有时候需要将HDFS或Hive上的数据导出到传统关系型数据库中（如MySQL、Oracle等），或者将传统关系型数据库中的数据导入到HDFS或Hive上，如果通过人工手动进行数据迁移的话，就会显得非常麻烦。为此，可使用Apache提供的Sqoop工具进行数据迁移。

一、Sqoop概述

1. Sqoop简介

Sqoop(发音：skup)是一款开源的工具，主要用于在Hadoop(Hive)与传统的数据库(mysql、postgresql…)间进行数据的传递，可以将一个关系型数据库（例如 ： MySQL ,Oracle ,Postgres等）中的数据导进到Hadoop的HDFS中，也可以将HDFS的数据导进到关系型数据库中。
Sqoop项目开始于2009年，最早是作为Hadoop的一个第三方模块存在，后来为了让使用者能够快速部署，也为了让开发人员能够更快速的迭代开发，Sqoop独立成为一个Apache项目。

2. Sqoop原理

Sqoop是传统关系型数据库服务器与Hadoop间进行数据同步的工具，其底层利用MapReduce并行计算模型以批处理方式加快数据传输速度，并且具有较好的容错性功能，工作流程如下所示。

Sqoop是关系型数据库与Hadoop之间的数据桥梁，这个桥梁的重要组件是Sqoop连接器，它用于实现与各种关系型数据库的连接，从而实现数据的导入和导出操作。

1) 导入原理

在导入数据之前，Sqoop使用JDBC检查导入的数据表，检索出表中的所有列以及列的SQL数据类型，并将这些SQL类型映射为Java数据类型，在转换后的MapReduce应用中使用这些对应的Java类型来保存字段的值，Sqoop的代码生成器使用这些信息来创建对应表的类，用于保存从表中抽取的记录。

2) 导出原理

在导出数据前，Sqoop会根据目标表的定义生成一个Java类，这个生成的类能够从文本中解析出记录数据，并能够向表中插入类型合适的值，然后启动一个MapReduce作业，从HDFS中读取源数据文件，使用生成的类解析出记录，并且执行选定的导出方法。

二、安装配置

1.下载

2.上传服务器

3. 解压

4. 配置sqoop

5. 配置环境变量并使其起作用

#Sqoop配置
export SQOOP_HOME=/export/servers/sqoop-1.4.7
export PATH=$PATH:$PATH:SQOOP_HOME/bin

6. 测试

查看版本

连接mysql

sqoop list-databases --connect jdbc:mysql://localhost:3306 --username root --password Bigdata20!

需要上传mysql的驱动包

重新运行命令

三、Sqoop指令介绍

Sqoop作为一款工具，开发者只需掌握工具的使用方式，它提供了一系列的工具指令，来进行数据的导入、导出操作等，开发人员只需输入”sqoop help“帮助指令查看帮助文档，如下所示。

上图就是 Sqoop 支持的所有工具命令，并且对应有英文解释说明。其中，包含了常用的导入（import）、导出（export）、显示所有数据库名称（list-databases）和显示所有表（list-tables）等。

通过 sqoop help command 可以查看该指令的各种参数，比如查看 import 命令的参数，可以使用 sqoop help import

注意：在执行 Sqoop 指令进行操作时可以指定 通用参数（Common arguments）和 特定参数 。通用参数主要用于对关系数据库的连接配置，而特定参数主要用于对 Sqoop 的具体操作实现进行功能配置，并且通用参数必须位于特定参数之前。

四、sqoop数据导入

Sqoop 数据导入（import）是将关系数据库中的单个表数据导入到 HDFS 和 Hive 等具有 Hadoop 分布式存储结构的文件系统中，表中的每一行都视为一条记录，所有记录默认以文本文件格式进行逐行储存，还可以以二进制行书储存，如 Avro 文件格式和序列文件格式（SequenceFile）。

下面来演示 Sqoop 数据导入、导出的相关操作：

1.数据准备

1) 启动并登录MySQL

2) 新建数据库并使用

首先在hadoop001机器上安装的数据MySQL数据库中创建 userdb 数据库，字符集设置为UTF-8：

create database userdb character set utf8 collate utf8_general_ci;

使用 userdb 数据库

use userdb;

3) 创建表并导入数据

下面是创建数据库表的脚本：

#------------------------------Table structure for `emp`----------------------------
DROP TABLE IF EXISTS `emp`;
CREATE TABLE `emp`
(
    `id`     int(11) NOT NULL,
    `name`   varchar(100) DEFAULT NULL,
    `deg`    varchar(100) DEFAULT NULL,
    `salary` int(11) DEFAULT NULL,
    `dept`   varchar(10)  DEFAULT NULL,
    PRIMARY KEY (`id`)
)
charset utf8 collate utf8_general_ci;
#------------------------------Records of emp----------------------------
INSERT INTO `emp`
VALUES ('1201', 'gopal', 'manager', '50000', 'TP');
INSERT INTO `emp`
VALUES ('1202', 'manisha', 'Proof reader', '50000', 'TP');
INSERT INTO `emp`
VALUES ('1203', 'khalil', 'php dev', '30000', 'AC');
INSERT INTO `emp`
VALUES ('1204', 'prasanth', 'php dev', '30000', 'AC');
INSERT INTO `emp`
VALUES ('1205', 'kranthi', 'admin', '20000', 'TP');

#------------------------------Table structrue for `emp_add`----------------------------
DROP TABLE IF EXISTS `emp_add`;
CREATE TABLE `emp_add`
(
    `id`     int(11) NOT NULL,
    `hno`    varchar(100) DEFAULT NULL,
    `street` varchar(100) DEFAULT NULL,
    `city`   varchar(100) DEFAULT NULL,
    PRIMARY KEY (`id`)
)
charset utf8 collate utf8_general_ci;
#------------------------------Records of emp_add----------------------------
INSERT INTO `emp_add`
VALUES ('1201', '288A', 'vgiri', 'jublee');
INSERT INTO `emp_add`
VALUES ('1202', '108I', 'aoc', 'sec-bad');
INSERT INTO `emp_add`
VALUES ('1203', '144Z', 'pgutta', 'hyd');
INSERT INTO `emp_add`
VALUES ('1204', '78B', 'old city', 'sec-bad');
INSERT INTO `emp_add`
VALUES ('1205', '720X', 'hitec', 'sec-bad');

#------------------------------Table structrue for `emp_conn`----------------------------
DROP TABLE IF EXISTS `emp_conn`;
CREATE TABLE `emp_conn`
(
    `id`    int(100) NOT NULL,
    `phno`  varchar(100) DEFAULT NULL,
    `email` varchar(100) DEFAULT NULL,
    PRIMARY KEY (`id`)
)
charset utf8 collate utf8_general_ci;
#------------------------------Records of emp_conn----------------------------
INSERT INTO `emp_conn`
VALUES ('1201', '2356742', 'gopal@tp.com');
INSERT INTO `emp_conn`
VALUES ('1202', '1661663', 'manisha@tp.com');
INSERT INTO `emp_conn`
VALUES ('1203', '8887776', 'khalil@ac.com');
INSERT INTO `emp_conn`
VALUES ('1204', '9988774', 'prasanth@ac.com');
INSERT INTO `emp_conn`
VALUES ('1205', '1231231', 'kranthi@tp.com');

这里有三种创建方法：

• 第一种，直接在MySQL命令窗口输入命令；
• 第二种，创建userdb.sql文件，使用source命令创建；
• 第三种，Navicat远程连接数据库，并新建查询，执行脚本；

这里我使用的是第三种方法：

创建完后：

2.启动hadoop集群

3.MySQL表数据导入HDFS（出现错误）

将 MySQL 表数据导入到 HDFS 中，具体指令示例如下（“ \ ”符号用于单个指令换行）。

bin/sqoop import \
--connect jdbc:mysql://hadoop001:3306/userdb \
--username root \
--password Bigdata20! \
--target-dir /sqoopresult \
--table emp \
--num-mappers 1

上述指令演示了将MySQL表数据导入到HDFS的基本使用，其中包含了多个参数，下面对其中的参数进行具体说明。

• –-connect 指定连接的关系数据库，包括 JDBC 驱动名、主机名、端口号和数据库名称。应注意的是，Sqoop 数据导入导出操作需要启动 Hadoop 集群的 MapReduce程序，所以这里应该注意的是，Sqoop 数据导入导出操作需要启动 Hadoop 集群的 MapReduce 程序，所以这里连接的主机名不能是 localhost，必须是 MySQL 数据库所在的主机名或 IP 地址。
• –-username 用于指定连接数据库的用户名
• –-password 用于指定连接数据库的密码。这种方式直接暴露了数据库连接密码，不太安全，所以可以使用 -P 指令代替，这个指令会以交互方式提示用户输入密码。
• –-target-dir 指定导入到 HDFS 的目录，代表 MySQL 数据表要导入 HDFS 的目标地址。这里需要注意该选项所指定的目录的最后一个子目录不能存在，否则 Sqoop 会执行失败。
• –-table 代表要进行数据导入操作的 MySQL 源数据库表名
• –-num-mappers 指定map任务个数（默认为4个，并且会产生4个结果文件），可简写为-m。这里指定 map 任务个数为1，那么只会启动一个 Map 程序执行相关操作，并只会生成一个结果文件

出现错误，root用户l权限不够：

4.使用允许远程连接的用户

bin/sqoop import --connect jdbc:mysql://hadoop001:3306/userdb --username 20bigdata --password Bigdata2022! --target-dir /sqoopresult --table emp --num-mappers 1 

注意：添加允许远程连接的用户

5.运行成功

五、增量导入

当MySQL表中的数据发生新增或修改变化，需要更新HDFS上对应的数据时，就可以使用Sqoop的增量导入功能，Sqoop目前支持两种增量导入模式：append模式和lastmodified模式。其中，append模式主要针对INSERT新增数据的增量导入； lastmodified模式主要针对UPDATE修改数据的增量导入。

在进行增量导入操作时，首先必须指定“–check-column”参数，用来检查数据表列字段，从而确定哪些数据需要执行增量导入。例如，在执行 append 模式增量导入时，通常会将“–check-column”参数指定为具有连续自增功能的列（如主键id）；而执行 lastmodified 模式增量导入时通常会将“–check-column”参数必须指定为日期时间类型的列（如 date 或 timestamp 类型的列）。

同时，还可以为增量导入操作指定“–last-value”参数，只用于增量导入 last-value 值以后的记录数据，然后存储到之前 HDFS 上相应目录下的一个单独文件中。

1. 向emp表中插入数据

先打开Navicat查看emp表

向emp表添加新数据，指令如下所示。

INSERT INTO `emp` VALUES ('1206','itcast','java dev','50000','AC');

2. 启动hadoop集群

因为前面已经启动过了，这里不在赘述。

3. 执行增量导入

然后，针对 emp 表数据的新增变化执行 append 模式的增量导入，指令如下。

bin/sqoop import --connect jdbc:mysql://hadoop001:3306/userdb --username 20bigdata --password Bigdata2022! --target-dir /sqoopresult --table emp --num-mappers 1 --incremental append --check-column id --last-value 1205 

4. 查看结果

六、sqoop数据导出

Sqoop导出与导入是相反的操作，也就是将HDFS、Hive、Hbase等文件系统或数据仓库中的数据导出到关系型数据库中，在导出操作之前，目标表必须存在于目标数据库中，否则在执行导出操作时会失败。

1. 创建MySql数据库

2. 创建表

use sqoop_migrate;
CREATE TABLE `employee` (
  `e_id` varchar(20) NOT NULL DEFAULT '',
  `e_name` varchar(20) NOT NULL DEFAULT '',
  `e_birth` varchar(20) NOT NULL DEFAULT '',
  `e_sex` varchar(10) NOT NULL DEFAULT '',
  PRIMARY KEY (`e_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8; 

3. 启动hadoop集群

4.创建文件并上传到hadoop集群上

6,测试用户6,2019-08-10,男
7,测试用户7,2019-08-11,男
8,测试用户8,2019-08-12,男
9,测试用户9,2019-08-13,女
10,测试用户10,2019-08-14,女

5. sqoop导出文件

将 HDFS 上/sqoopresult 目录下的 part-m-00000 文件进行导出操作，指令如下：

sqoop export --connect jdbc:mysql://hadoop001:3306/sqoop_migrate?characterEncoding=utf8 \
--username '20bigdata' \
--password 'Bigdata2022!'  \
--num-mappers 1  \
--table employee  \
--columns "e_id,e_name,e_birth,e_sex"  \
--export-dir '/sqoopresult/mysqltest.txt'  \
--fields-terminated-by ',' 

执行成功：

6.查看结果

参考文章：

http://t.csdn.cn/rk3KG

http://t.csdn.cn/hB24i