数据导入导出工具Sqoop入门

2022-11-13  本文已影响0人  文景大大

一、Sqoop是什么

Apache Sqoop是Hadoop生态体系和RDBMS体系之间相互传输数据的一种工具，其工作机制是将导入、导出命令翻译为MapReduce程序运行，使得HDFS、Hive、Hbase中的数据能和MySQL、Oracle、DB2之类的关系型数据库中的数据进行双向流动。

sqoop工作示意图

从Apache的官网可以看到Sqoop目前已经被移到Attic中了，意味着该项目的生命周期也结束了，后续不会再有新的版本发布或者旧版本的维护和更新了，其实Sqoop已经比较成熟，也没什么需要增加的新特性了，所以仍然可以放心使用。

二、Sqoop的安装

Sqoop主要分为1.x和2.x，前者1.x是使用最多的，比较成熟稳定，最后更新版本为1.4.7；而后者2.x并没有流行开来，使用上相比前者也没有太多的亮点，而且不兼容1.x，因此本文主要学习1.x版本。

首先确保服务器上拥有了Java和健康的Hadoop集群环境；

然后从官网下载安装包：Index of /dist/sqoop/1.4.6 (apache.org)，放到对应的服务器上，并解压缩。为什么不使用最新版1.4.7？参考6.1节内容。

然后修改sqoop的配置文件：

cd sqoop/conf
mv sqoop-env-template.sh sqoop-env.sh
vi sqoop-env.sh

#修改内容如下
export HADOOP_COMMON_HOME=/usr/service/hadoop/hadoop-3.2.2
export HADOOP_MAPRED_HOME=/usr/service/hadoop/hadoop-3.2.2
export HIVE_HOME=/usr/service/hive/apache-hive-3.1.2-bin
#仅演示hdfs、hive配合sqoop的使用，其它的暂不配置

然后将关系型数据库的Java驱动包放到sqoop的lib目录下，这里以MySQL作为演示：

cd ~
cp ./mysql-connector-java-5.1.32.jar ./sqoop/lib/

然后，我们添加Sqoop的环境变量：

vi /etc/profile
export SQOOP_HOME=/root/zx-test/sqoop
export PATH=$PATH:$SQOOP_HOME/bin
export CLASSPATH=$CLASSPATH:$SQOOP_HOME/lib
source /etc/profile

到此我们就完成了sqoop的配置，然后需要找一个MySQL服务，然后测试一下连接：

cd ./sqoop/bin
sqoop list-databases --connect jdbc:mysql://10.x.x.x:3306/ --username root --password yourpassword
#正常情况下会输出该数据库里面所有的数据库名称

三、使用Sqoop导入数据

首先，我们在MySQL数据库中准备一些数据：

DROP TABLE IF EXISTS `emp`;
CREATE TABLE `emp` (
  `id` int(11) DEFAULT NULL,
  `name` varchar(100) DEFAULT NULL,
  `deg` varchar(100) DEFAULT NULL,
  `salary` int(11) DEFAULT NULL,
  `dept` varchar(10) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

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');

3.1 全量导入到HDFS

sqoop import \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
# 指定上传到HDFS的目录
--target-dir /zx/sqoop \
# 如上述目录已经存在则删除（可选）
--delete-target-dir \
# 指定导入MySQL中的哪张表
--table emp \
# 指定导入的并发度，并发度n表示并发导入，HDFS中会存在n个导入文件（可选，默认为1）
--m 1

然后执行完毕后查看hdfs中的结果：

hdfs dfs -cat /zx/sqoop/part-m-00000
1201,gopal,manager,50000,TP
1202,manisha,Proof reader,50000,TP
1203,khalil,php dev,30000,AC
1204,prasanth,php dev,30000,AC
1205,kranthi,admin,20000,TP

发现和MySQL中emp表中的一致，全量导入成功了。

3.2 全量导入到Hive

因为Hive是有表结构的，所以我们有两种方式：

导入到HDFS和Hive命令基本差不多，后面默认就以导入到HDFS来作为演示了。

3.3 导入子集

有些时候，我们只需要MySQL中全量数据中的部分符合条件的数据导入Hadoop，那么就可以使用where条件进行过滤：

sqoop import \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
--target-dir /zx/sqoop \
--delete-target-dir \
--where "salary >= 30000" \
--table emp \
--m 1

任务完成后，我们可以在HDFS对应目录下查看结果，看看其中的内容是不是对salary做了过滤的：

hdfs dfs -cat /zx/sqoop/part-m-00000                                                 
1201,gopal,manager,50000,TP
1202,manisha,Proof reader,50000,TP
1203,khalil,php dev,30000,AC
1204,prasanth,php dev,30000,AC

还有些场景，我们只需要MySQL中全量数据中的部分字段导入Hadoop，那么就可以使用query sql语句来进行筛选：

sqoop import \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
--target-dir /zx/sqoop \
--delete-target-dir \
--query 'select id,name,salary from emp where salary >= 30000 and $CONDITIONS' \
# 指定导入并发为2
--m 2 \
# 指定根据emp表中的id字段来并发
--split-by id \
# 指定导入HDFS中字段的分隔符，默认为英文逗号，此处指定为制表符
--fields-terminated-by '\t'

任务完成后，我们可以在HDFS对应目录下看到两个文件part-m-00000和part-m-00001，可以看看其中的内容是不是对salary做了过滤的：

hdfs dfs -cat /zx/sqoop/part-m-00000                                                 
1201    gopal   50000
1202    manisha 50000
hdfs dfs -cat /zx/sqoop/part-m-00001
1203    khalil  30000
1204    prasanth        30000

3.4 Append模式增量导入

首先我们3.1节介绍全量导入的时候已经将emp中的5条数据都导入到HDFS中了，现在我们在MySQL中再添加两条数据：

insert into `emp` (`id`, `name`, `deg`, `salary`, `dept`) values ('1206', 'allen', 'admin', '30000', 'tp');
insert into `emp` (`id`, `name`, `deg`, `salary`, `dept`) values ('1207', 'woon', 'admin', '40000', 'tp');

此时如何以增量的方式只将增加的数据导入HDFS呢？

sqoop import \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
--target-dir /zx/sqoop \
--table emp \
#使用append追加模式，对大于等于last-value指定值之后的记录进行追加导入
--incremental append \
#指定列用来判断是否是增量数据，该列不能是字符、字符串类型，最好是能线性增长的自增字段或者时间戳类型
--check-column id \
#指定check-column字段上次导入的最大值
--last-value 1205 \
--m 1

任务结束后查看结果：

hdfs dfs -cat /zx/sqoop/part-m-00000                                                 
1201,gopal,manager,50000,TP
1202,manisha,Proof reader,50000,TP
1203,khalil,php dev,30000,AC
1204,prasanth,php dev,30000,AC
1205,kranthi,admin,20000,TP
hdfs dfs -cat /zx/sqoop/part-m-00001
1206,allen,admin,30000,tp
1207,woon,admin,40000,tp

3.5 LastModefied模式增量导入

如上append模式只能对新增加的记录进行增量导入，加入先前已经导入的数据发生了改变该怎么办呢？我们希望修改的记录也能导入，就要使用增量导入的LastModefied模式。首先我们重新创建emp表，增加修改时间字段，这个字段在记录新增或者修改时会自动更新。

CREATE TABLE `emp` (
  `id` int(11) DEFAULT NULL,
  `name` varchar(100) DEFAULT NULL,
  `deg` varchar(100) DEFAULT NULL,
  `salary` int(11) DEFAULT NULL,
  `dept` varchar(10) DEFAULT null,
  `last_mod` timestamp default current_timestamp on update current_timestamp
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

INSERT INTO emp (id,name,deg,salary,dept) VALUES
     (1201,'gopal','manager',50000,'TP'),
     (1202,'manisha','Proof reader',50000,'TP'),
     (1203,'khalil','php dev',30000,'AC'),
     (1204,'prasanth','php dev',30000,'AC'),
     (1205,'kranthi','admin',20000,'TP');

然后我们将如上5条数据导入HDFS，参见3.1，然后我们执行新增和修改记录：

INSERT INTO emp (id,name,deg,salary,dept) VALUES
     (1206,'allen','admin',30000,'tp');
update emp set salary = 1000 where id = 1201;

然后执行LastModefied模式增量导入：

sqoop import \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
--target-dir /zx/sqoop \
--table emp \
#使用lastmodified追加模式
--incremental lastmodified \
#指定lastmodified对应的字段
--check-column last_mod \
#指定最后一次追加之后的一个时间，注意必须大于最后一次追加的时间，不然最后一次追加的记录会被重复导入
--last-value "2022-11-13 15:00:42" \
#以追加文件的方式导入
--append \
--m 1

可以看到新增加的导入文件中，插入和更新的记录都有了：

hdfs dfs -cat /zx/sqoop/part-m-00000                                                 
1201,gopal,manager,50000,TP,2022-11-13 14:59:46.0
1202,manisha,Proof reader,50000,TP,2022-11-13 14:59:46.0
1203,khalil,php dev,30000,AC,2022-11-13 14:59:46.0
1204,prasanth,php dev,30000,AC,2022-11-13 14:59:46.0
1205,kranthi,admin,20000,TP,2022-11-13 14:59:46.0
hdfs dfs -cat /zx/sqoop/part-m-00001
1201,gopal,manager,1000,TP,2022-11-13 15:06:25.0
1206,allen,admin,30000,tp,2022-11-13 15:05:42.0

但是有个问题，有的时候，我们并不想要多个文件来存储，能否把这些增量数据对应的文件和第一次全量导入对应的文件合并起来呢？使用merge-key就可以了：

INSERT INTO emp (id,name,deg,salary,dept) VALUES
     (1207,'woon','admin',40000,'tp');

update emp set salary = 90000 where id = 1201;
update emp set salary = 10000 where id = 1206;

然后执行导入操作：

sqoop import \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
--target-dir /zx/sqoop \
--table emp \
#使用lastmodified追加模式
--incremental lastmodified \
#指定lastmodified对应的字段
--check-column last_mod \
#指定最后一次追加之后的一个时间，注意必须大于最后一次追加的时间，不然最后一次追加的记录会被重复导入
--last-value "2022-11-13 15:00:42" \
#以merge-key的形式导入，只会有一个文件
--merge-key id \
--m 1

此时，sqoop会执行两个操作，首先将增量数据导入HDFS中，然后将所有的文件按照merge-key进行合并成为一个文件：

hdfs dfs -cat /zx/sqoop/part-r-00000
1201,gopal,manager,90000,TP,2022-11-13 15:17:16.0
1202,manisha,Proof reader,50000,TP,2022-11-13 14:59:46.0
1203,khalil,php dev,30000,AC,2022-11-13 14:59:46.0
1204,prasanth,php dev,30000,AC,2022-11-13 14:59:46.0
1205,kranthi,admin,20000,TP,2022-11-13 14:59:46.0
1206,allen,admin,10000,tp,2022-11-13 15:17:33.0
1207,woon,admin,40000,tp,2022-11-13 15:17:14.0

可以看到，上面part-m-00000和part-m-00001都被合并为了part-r-00000

四、使用Sqoop导出数据

导出数据到关系型数据的前提是，目标表必须是已经存在了的。

4.1 插入导出模式

默认情况下，sqoop是将需要导入的数据转换为一个个的insert语句添加到目标数据库表中，sqoop无法检查该表是否具有约束，所以需要使用者自己避免表存在约束从而导致sqoop导入失败的情况。因此，该默认模式最常用于将数据导入空表中的场景。

首先我们准备emp_data.txt文件内容如下：

1201,gopal,manager,50000,TP
1202,manisha,preader,50000,TP
1203,kalil,php dev,30000,AC
1204,prasanth,php dev,30000,AC
1205,kranthi,admin,20000,TP
1206,satishp,grpdes,20000,GR

然后将其上传到HDFS上：

hdfs dfs -mkdir /zx/sqoop
hdfs dfs -put emp_data.txt /zx/sqoop

然后在目标数据库中新建表：

create table emp ( 
   id int not null primary key, 
   name varchar(20), 
   deg varchar(20),
   salary int,
   dept varchar(10));

然后就可以执行导入命令了：

sqoop export \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
#要导入的数据库表名称
--table emp \
#HDFS中文件的路径
--export-dir /zx/sqoop/emp_data.txt \
#指定文件中字段之间的分隔符，默认就是英文逗号（可选）
--input-fields-terminated-by ',' \
#选择目标表中的列并控制它们的顺序，当HDFS文件中的字段顺序和表字段顺序一致时可以不写（可选）
--columns id,name,deg,salary,dept

4.2 更新导出模式

上述插入导出模式适用于全量导出到空表的场景，但是更多的场景需要我们进行增量导出，比如修改了某条记录的值，那么就需要使用更新导出模式：

我们假设要修改id为1201和1204的emp的薪水为10000，那么我们准备一个新的数据文件emp_data_02.txt：

1201,gopal,manager,10000,TP
1204,prasanth,php dev,10000,AC
1207,jack,python,32000,TG

其中第1、2条是更新，第3条是新增。然后将其导入到HDFS上：

hdfs dfs -put emp_data_01.txt /zx/sqoop/

然后执行增量导入命令：

sqoop export \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
--password yourpassword \
--table emp \
--export-dir /zx/sqoop/emp_data_01.txt \
#指定根据哪个字段来进行更新，可以同时指定多个字段，用英文逗号隔开
--update-key id \
#指定仅更新已存在的记录，不会插入新的记录（默认）
--update-mode updateonly

当然了，如果新增的记录也想要导入，那么update-mode修改为allowinsert即可。

五、Sqoop作业的使用

Job的出现可以将某一项日常任务设为一个工作模板，使得我们每天执行该任务的时候不用像之前一样需要敲很多命令，特别是其中还包含了数据库的敏感信息。同时Job还可以配合其它任务调度框架使得Job能够自动化执行，是非常便捷和高效的。

此处我们选择将3.5中的例子作为一个JOB来执行，首先我们准备一张数据库表：

create table emp ( 
   id int not null primary key, 
   name varchar(20), 
   deg varchar(20),
   salary int,
   dept varchar(10),
  last_mod timestamp default current_timestamp on update current_timestamp);
  
INSERT INTO `emp` (`id`, `name`, `deg`, `salary`, `dept`) VALUES ('1201', 'gopal', 'manager', '50000', 'TP');
INSERT INTO `emp` (`id`, `name`, `deg`, `salary`, `dept`) VALUES ('1202', 'manisha', 'Proof reader', '50000', 'TP');
INSERT INTO `emp` (`id`, `name`, `deg`, `salary`, `dept`) VALUES ('1203', 'khalil', 'php dev', '30000', 'AC');
INSERT INTO `emp` (`id`, `name`, `deg`, `salary`, `dept`) VALUES ('1204', 'prasanth', 'php dev', '30000', 'AC');
INSERT INTO `emp` (`id`, `name`, `deg`, `salary`, `dept`) VALUES ('1205', 'kranthi', 'admin', '20000', 'TP');

5.1设置免密执行

方式一：修改Sqoop的配置文件sqoop-site.xml的如下内容：

  <property>
    <name>sqoop.metastore.client.autoconnect.password</name>
    <value>true</value>
    <description>The password to bind to the metastore.
    </description>
  </property>

如此第一次手动输入密码后就会将密码记录到元数据里面，下次启动JOB就不需要再输入密码了，不建议使用这种方式。

方式二：创建一个专门存储密码的文件zhangsan_db.pwd，并将其上传到HDFS上，也可以是服务器上的本地文件。

yourpassword

hdfs dfs -mkdir /zx/sqoop
hdfs dfs -put zhangsan_db.pwd /zx/sqoop

5.2 创建JOB

随后我们就可以开始创建该Job了，这里使用的是3.5中merge-key模式的增量导入：

#注意import前面有空格
sqoop job --create update_emp_job -- import \
--connect jdbc:mysql://10.x.x.x:3306/zhangsan \
--username root \
#指定使用的密码存放文件
--password-file /zx/sqoop/zhangsan_db.pwd \
--target-dir /zx/sqoop \
--table emp \
--incremental lastmodified \
--check-column last_mod \
--last-value "2022-11-13 15:00:42" \
--merge-key id \
--m 1

5.3 查看JOB

#查看所有的JOB列表
sqoop job --list

Available jobs:
  update_emp_job
  
#查看具体某个JOB的信息
sqoop job --show update_emp_job
Job: update_emp_job
Tool: import
Options:
----------------------------
verbose = false
incremental.last.value = 2022-11-13 15:00:42
db.connect.string = jdbc:mysql://10.x.x.x:3306/zhangsan
codegen.output.delimiters.escape = 0
codegen.output.delimiters.enclose.required = false
codegen.input.delimiters.field = 0
hbase.create.table = false
hdfs.append.dir = false
db.table = emp
codegen.input.delimiters.escape = 0
import.fetch.size = null
accumulo.create.table = false
codegen.input.delimiters.enclose.required = false
db.username = root
reset.onemapper = false
codegen.output.delimiters.record = 10
import.max.inline.lob.size = 16777216
hbase.bulk.load.enabled = false
hcatalog.create.table = false
db.clear.staging.table = false
incremental.col = last_mod
codegen.input.delimiters.record = 0
db.password.file = /zx/sqoop/zhangsan_db.pwd
enable.compression = false
hive.overwrite.table = false
hive.import = false
codegen.input.delimiters.enclose = 0
accumulo.batch.size = 10240000
hive.drop.delims = false
codegen.output.delimiters.enclose = 0
hdfs.delete-target.dir = false
codegen.output.dir = .
codegen.auto.compile.dir = true
relaxed.isolation = false
mapreduce.num.mappers = 1
accumulo.max.latency = 5000
import.direct.split.size = 0
codegen.output.delimiters.field = 44
export.new.update = UpdateOnly
incremental.mode = DateLastModified
hdfs.file.format = TextFile
codegen.compile.dir = /tmp/sqoop-root/compile/0f4dec925023129aff3a49ff0c7f9159
direct.import = false
hdfs.target.dir = /zx/sqoop
hive.fail.table.exists = false
merge.key.col = id
db.batch = false

5.4 运行JOB

sqoop job --exec update_emp_job

后续可以将该指令设为Linux定时任务，或者用Oozie、Azkaban等任务调度框架去定时去执行它，解放了双手。

为什么你执行JOB会报错：Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/commons/lang/StringUtils，参考6.2节内容。

为什么你的数据库密码会无法识别：Error: java.lang.RuntimeException: Can't parse input data: 'yourpassword'，参考6.3节内容。

六、备注防坑：

（1）起初是使用1.4.7版本作为实验的，但是sqoop命令执行总是报错如下：

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/commons/lang/StringUtils
        at org.apache.sqoop.manager.MySQLManager.initOptionDefaults(MySQLManager.java:73)
        at org.apache.sqoop.manager.SqlManager.<init>(SqlManager.java:89)
        at com.cloudera.sqoop.manager.SqlManager.<init>(SqlManager.java:33)
        at org.apache.sqoop.manager.GenericJdbcManager.<init>(GenericJdbcManager.java:51)
        at com.cloudera.sqoop.manager.GenericJdbcManager.<init>(GenericJdbcManager.java:30)
        at org.apache.sqoop.manager.CatalogQueryManager.<init>(CatalogQueryManager.java:46)
        at com.cloudera.sqoop.manager.CatalogQueryManager.<init>(CatalogQueryManager.java:31)
        at org.apache.sqoop.manager.InformationSchemaManager.<init>(InformationSchemaManager.java:38)
        at com.cloudera.sqoop.manager.InformationSchemaManager.<init>(InformationSchemaManager.java:31)
        at org.apache.sqoop.manager.MySQLManager.<init>(MySQLManager.java:65)
        at org.apache.sqoop.manager.DefaultManagerFactory.accept(DefaultManagerFactory.java:67)
        at org.apache.sqoop.ConnFactory.getManager(ConnFactory.java:184)
        at org.apache.sqoop.tool.BaseSqoopTool.init(BaseSqoopTool.java:272)
        at org.apache.sqoop.tool.ListDatabasesTool.run(ListDatabasesTool.java:44)
        at org.apache.sqoop.Sqoop.run(Sqoop.java:147)
        at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:76)
        at org.apache.sqoop.Sqoop.runSqoop(Sqoop.java:183)
        at org.apache.sqoop.Sqoop.runTool(Sqoop.java:234)
        at org.apache.sqoop.Sqoop.runTool(Sqoop.java:243)
        at org.apache.sqoop.Sqoop.main(Sqoop.java:252)
Caused by: java.lang.ClassNotFoundException: org.apache.commons.lang.StringUtils
        at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:355)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
        ... 20 more

网上找到的教程是说1.4.7版本缺少依赖包，需要自己下载：Sqoop安装配置_Siobhan. 明鑫的博客-CSDN博客_sqoop安装与配置，自己嫌麻烦，就用了1.4.6老的版本继续实验，发现没有问题。

（2）即使使用了1.4.6版本，在执行sqoop的JOB的时候，还是会报错（1）中的内容，解决办法就是去apache官网下载common-lang的jar包，并把该包放到sqoop的lib目录下就行了，注意一定要下载2.x版本，3.x版本不管用。

（3）这个问题花了很久找寻和实验，一直没有解决，知道的朋友不妨告知一下。

上一篇下一篇

猜你喜欢

热点阅读