[大数据] Sqoop的安装及使用Sqoop导入至mysql、hive、hbase

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Sqoop的安装及使用Sqoop导入至mysql、hive、hbase -> 正文阅读

[大数据]Sqoop的安装及使用Sqoop导入至mysql、hive、hbase

1、Sqoop的安装

解压sqoop安装包

tar -zxvf /opt/download/sqoop-1.4.6.bin__hadoop-2.0.4-alpha.tar.gz -C /opt/software/

改名

 mv /opt/software/sqoop-1.4.6.bin__hadoop-2.0.4-alpha/ /opt/software/sqoop146/

复制出配置文件副本，并改名

cp /opt/software/sqoop146/conf/sqoop-env-template.sh /opt/software/sqoop146/conf/sqoop-env.sh

打开配置文件

vim /opt/software/sqoop146/conf/sqoop-env.sh

添加配置信息

export HADOOP_COMMON_HOME=/opt/software/hadoop313

export HADOOP_MAPRED_HOME=/opt/software/hadoop313

export HBASE_HOME=/opt/software/hbase235

export HIVE_HOME=/opt/software/hive312

export ZOOCFGDIR=/opt/software/zookeeper357

在这里插入图片描述

将mysql和hadoop驱动包放置lib目录下

cp /opt/software/hive312/lib/mysql-connector-java-5.1.47.jar /opt/software/sqoop146/lib/ ./

cp /opt/software/hadoop/hadoop313/share/hadoop/common/hadoop-common-3.1.3.jar ./
cp /opt/software/hadoop/hadoop313/share/hadoop/hdfs/hadoop-hdfs-3.1.3.jar ./
cp /opt/software/hadoop/hadoop313/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.1.3.jar ./

添加全局环境变量

vim /etc/profile.d/myenv.sh#sqoop

################################
export SQOOP_HOME=/opt/software/sqoop146
export PATH=$PATH:$SQOOP_HOME/bin
export LOGDIR=$SQOOP_HOME/logs/
###############################

#激活环境变量
source /etc/profile

测试能否成功

sqoop list-databases --connect jdbc:mysql://你的ip地址:3306 --username 用户名 --password 密码
sqoop list-databases --connect jdbc:mysql://192.168.158.202:3306 --username root --password 12345678

2、Sqoop的使用

1.Sqoop的常用命令

	命令	类	说明
1	import	ImportTool	将数据导入到集群
2	export	ExportTool	将集群数据导出
3	codegen	CodeGenTool	获取数据库中某张表数据生成Java并打包Jar
4	create-hive-table	CreateHiveTableTool	创建Hive表
5	eval	EvalSqlTool	查看SQL执行结果
6	import-all-tables	ImportAllTablesTool	导入某个数据库下所有表到HDFS中
7	job	JobTool	用来生成一个sqoop的任务，生成后，该任务并不执行，除非使用命令执行该任务。
8	list-databases	ListDatabasesTool	列出所有数据库名
9	list-tables	ListTablesTool	列出某个数据库下所有表
10	merge	MergeTool	将HDFS中不同目录下面的数据合在一起，并存放在指定的目录中
11	metastore	MetastoreTool	记录sqoop job的元数据信息，如果不启动metastore实例，则默认的元数据存储目录为：~/.sqoop，如果要更改存储目录，可以在配置文件sqoop-site.xml中进行更改。
12	help	HelpTool	打印sqoop帮助信息
13	version	VersionTool	打印sqoop版本信息

2.命令对应参数解析

2.1数据库连接参数

	参数	说明
1	–connect	连接关系型数据库的URL
2	–connection-manager	指定要使用的连接管理类
3	–driver	Hadoop根目录
4	–help	打印帮助信息
5	–password	连接数据库的密码
6	–username	连接数据库的用户名
7	–verbose	在控制台打印出详细信息

2.2 import参数

	参数	说明
1	–enclosed-by	给字段值前加上指定的字符
2	–escaped-by	对字段中的双引号加转义符
3	–fields-terminated-by	设定每个字段是以什么符号作为结束，默认为逗号
4	–lines-terminated-by	设定每行记录之间的分隔符，默认是\n
5	–mysql-delimiters	Mysql默认的分隔符设置，字段之间以逗号分隔，行之间以\n分隔，默认转义符是\，字段值以单引号包裹。
6	–optionally-enclosed-by	给带有双引号或单引号的字段值前后加上指定字符。

特有参数

	参数	说明
1	–append	将数据追加到HDFS中已经存在的DataSet中，如果使用该参数，sqoop会把数据先导入到临时文件目录，再合并。
2	–as-avrodatafile	将数据导入到一个Avro数据文件中
3	–as-sequencefile	将数据导入到一个sequence文件中
4	–as-textfile	将数据导入到一个普通文本文件中
5	–boundary-query	边界查询，导入的数据为该参数的值（一条sql语句）所执行的结果区间内的数据。
6	–columns <col1, col2, col3>	指定要导入的字段
7	–direct	直接导入模式，使用的是关系数据库自带的导入导出工具，以便加快导入导出过程。
8	–direct-split-size	在使用上面direct直接导入的基础上，对导入的流按字节分块，即达到该阈值就产生一个新的文件
9	–inline-lob-limit	设定大对象数据类型的最大值
10	–m或–num-mappers	启动N个map来并行导入数据，默认4个。
11	–query或–e	将查询结果的数据导入，使用时必须伴随参–target-dir，–hive-table，如果查询中有where条件，则条件后必须加上$CONDITIONS关键字
12	–split-by	按照某一列来切分表的工作单元，不能与–autoreset-to-one-mapper连用（请参考官方文档）
13	–table	关系数据库的表名
14	–target-dir	指定HDFS路径
15	–warehouse-dir	与14参数不能同时使用，导入数据到HDFS时指定的目录
16	–where	从关系数据库导入数据时的查询条件
17	–z或–compress	允许压缩
18	–compression-codec	指定hadoop压缩编码类，默认为gzip(Use Hadoop codec default gzip)
19	–null-string	string类型的列如果null，替换为指定字符串
20	–null-non-string	非string类型的列如果null，替换为指定字符串
21	–check-column < col >	作为增量导入判断的列名
22	–incremental	mode：append或lastmodified
23	–last-value	指定某一个值，用于标记增量导入的位置

2.3 export参数

	参数	说明
1	–input-enclosed-by	对字段值前后加上指定字符
2	–input-escaped-by	对含有转移符的字段做转义处理
3	–input-fields-terminated-by	字段之间的分隔符
4	–input-lines-terminated-by	行之间的分隔符
5	–input-optionally-enclosed-by	给带有双引号或单引号的字段前后加上指定字符

特有参数

	参数	说明
1	–direct	利用数据库自带的导入导出工具，以便于提高效率
2	–export-dir	存放数据的HDFS的源目录
3	-m或–num-mappers	启动N个map来并行导入数据，默认4个
4	–table	指定导出到哪个RDBMS中的表
5	–update-key	对某一列的字段进行更新操作
6	–update-mode	updateonlyallowinsert(默认)
7	–input-null-string	请参考import该类似参数说明
8	–input-null-non-string	请参考import该类似参数说明
9	–staging-table	创建一张临时表，用于存放所有事务的结果，然后将所有事务结果一次性导入到目标表中，防止错误。
10	–clear-staging-table	如果第9个参数非空，则可以在导出操作执行前，清空临时事务结果表

2.4 Hive参数

	参数	说明
1	–hive-delims-replacement	用自定义的字符串替换掉数据中的\r\n和\013 \010等字符
2	–hive-drop-import-delims	在导入数据到hive时，去掉数据中的\r\n\013\010这样的字符
3	–map-column-hive	生成hive表时，可以更改生成字段的数据类型
4	–hive-partition-key	创建分区，后面直接跟分区名，分区字段的默认类型为string
5	–hive-partition-value	导入数据时，指定某个分区的值
6	–hive-home	hive的安装目录，可以通过该参数覆盖之前默认配置的目录
7	–hive-import	将数据从关系数据库中导入到hive表中
8	–hive-overwrite	覆盖掉在hive表中已经存在的数据
9	–create-hive-table	默认是false，即，如果目标表已经存在了，那么创建任务失败。
10	–hive-table	后面接要创建的hive表,默认使用MySQL的表名
11	–table	指定关系数据库的表名

特有参数

	参数	说明
1	–hive-home	Hive的安装目录，可以通过该参数覆盖掉默认的Hive目录
2	–hive-overwrite	覆盖掉在Hive表中已经存在的数据
3	–create-hive-table	默认是false，如果目标表已经存在了，那么创建任务会失败
4	–hive-table	后面接要创建的hive表
5	–table	指定关系数据库的表名

2.5 Job参数

	参数	说明
1	–create	创建job参数
2	–delete	删除一个job
3	–exec	执行一个job
4	–help	显示job帮助
5	–list	显示job列表
6	–meta-connect	用来连接metastore服务
7	–show	显示一个job的信息
8	–verbose	打印命令运行时的详细信息

3、Sqoop常用操作

1.hdfs导入到mysql

#mysql
#--创建表
create table sqp_order(
create_date date,
user_name varchar(20),
total_volume decimal(10,2)
);
#--数据文件上传至hdfs
--orderinfo.log
-----------------------
2021-06-29,uzi,56.34
2021-06-29,ming,78.25
2021-06-29,xiaohu,67.09
2021-06-29,mlxg,23.67
2021-06-29,letme,21.45
-----------------------

#使用sqoop将hdfs数据导入到mysql
sqoop export \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table sqp_order \
-m 1 \
--export-dir /sqooptest/orderinfo \
--fields-terminated-by ',' \	
--lines-terminated-by '\n'

2.mysql导入到hdfs

全量导入

sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table sqp_order \
-m 1 \
--delete-target-dir \
--target-dir /sqooptest/mysql2hdfs \
--fields-terminated-by '\t' \
--lines-terminated-by '\n'

列剪裁

sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table sqp_order \
--columns user_name,total_volume \
-m 1 \
--delete-target-dir \
--target-dir /sqooptest/mysql2hdfsbyrow \
--fields-terminated-by '\t' \
--lines-terminated-by '\n'

行裁剪+多个reducer

sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table sqp_order \
--columns user_name,total_volume \
--where "total_volume<50" \
-m 2 \
--split-by user_name \
--delete-target-dir \
--target-dir /sqooptest/m2hbylinesplit \
--fields-terminated-by '\t' \
--lines-terminated-by '\n'

#另外一种写法
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--query "select user_name,total_volume from sqp_order where total_volume<50 and \$CONDITIONS" \
-m 2 \
--split-by user_name \
--delete-target-dir \
--target-dir /sqooptest/m2hbylinesplit2 \
--fields-terminated-by '\t' \
--lines-terminated-by '\n'

增量导入|append

#数据初始
create table add_test(
id int(4) not null,
name varchar(10) not null,
age int(3)
);
insert into add_test(id,name,age) values
(1,'xiaohu','18'),
(2,'wei','19'),
(3,'cryin','20'),
(4,'gala8','21'),
(5,'ming','22');

### 第一次导入 ###
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--query "select * from add_test where \$CONDITIONS" \
-m 1 \
--target-dir /sqooptest/m2hadd \
--fields-terminated-by '\t' \
--lines-terminated-by '\n' \
--check-column id \
--incremental append \
--last-value 0

---------------------
#运行结果
INFO tool.ImportTool:   --last-value 5
---------------------

insert into add_test(id,name,age) values
(6,'xiaohu1','18'),
(7,'wei1','19'),
(8,'cryin1','20'),
(9,'gala81','21'),
(10,'ming1','22');

### 增量导入 ###
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--query "select * from add_test where \$CONDITIONS" \
-m 1 \
--target-dir /sqooptest/m2hadd \
--fields-terminated-by '\t' \
--lines-terminated-by '\n' \
--check-column id \
--incremental append \
--last-value 5

---------------------
#运行结果
INFO tool.ImportTool:   --last-value 10
---------------------

增量导入|lastmodified

使用lastmodified方式导入数据要指定增量数据是要–append（追加）还是要–merge-key（合并）

#数据初始
create table add_test2(
Name varchar(20),
Times timestamp
);
insert into add_test2(Name,Times) values
('xiaohu','2021-06-29 13:23:31.0'),
('uzi','2021-06-29 14:23:31.0'),
('zzt','2021-06-29 15:23:31.0'),
('xlb','2021-06-29 16:23:31.0'),
('ming','2021-06-29 17:23:31.0');

### 第一次导入 ###
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--query "select * from add_test2 where \$CONDITIONS" \
-m 1 \
--target-dir /sqooptest/m2hadd2 \
--fields-terminated-by '\t' \
--lines-terminated-by '\n' \
--check-column Times \
--incremental lastmodified \
--last-value '0000-00-00 00:00:00'

---------------------
#运行结果
INFO tool.ImportTool:   --last-value 2021-07-04 16:59:35.0
---------------------

insert into add_test2(Name,Times) values
('xiaohu1','2021-07-04 17:01:31.0'),
('uzi1','2021-07-04 17:01:33.0'),
('zzt1','2021-07-04 17:02:35.0'),
('xlb1','2021-07-04 17:02:37.0'),
('ming1','2021-07-04 17:01:36.0');

### 增量导入 ###
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--query "select * from add_test2 where \$CONDITIONS" \
-m 1 \
--target-dir /sqooptest/m2hadd2 \
--fields-terminated-by '\t' \
--lines-terminated-by '\n' \
--check-column Times \
--incremental lastmodified \
--append \
--last-value '2021-07-04 16:59:35.0'

3.mysql导入到hive中

全量导入

sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table add_test2 \
-m 1 \
--hive-import \
--hive-table sqooptest.sqptest \
--create-hive-table 

-----------------------------
insert into add_test2(Name,Times) values
('xiaohu2','2021-07-03 17:01:31.0'),
('uzi2','2021-07-03 17:01:33.0'),
('zzt3','2021-07-03 17:02:35.0'),
('xlb4','2021-07-03 17:02:37.0'),
('ming5','2021-07-03 17:01:36.0');
-----------------------------
#追加
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table add_test2 \
--where "cast(Times as date)='2021-07-03'" \
-m 1 \
--hive-import \
--hive-table sqooptest.sqptest

间接导入

方式一

#hive中创建表
create table sqp_par(
Name varchar(20),
Times date
)
partitioned by (dates date)
row format delimited
fields terminated by ','
lines terminated by '\n'
stored as textfile;
#迁移数据
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table add_test2 \
--where "cast(Times as date)='2021-06-29'" \
-m 1 \
--delete-target-dir \
--target-dir /user/hive/warehouse/sqooptest.db/sqp_par/dates=2021-06-29 \
--fields-terminated-by ',' \
--lines-terminated-by '\n'
#添加分区
alter table sqp_par add partition(dates='2021-06-29');


#追加
sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table add_test2 \
--where "cast(Times as date)='2021-07-04'" \
-m 1 \
--delete-target-dir \
--target-dir /user/hive/warehouse/sqooptest.db/sqp_par/dates=2021-07-04 \
--fields-terminated-by ',' \
--lines-terminated-by '\n'

alter table sqp_par add partition(dates='2021-07-04');

方式二

create table sqp_par2(
Name varchar(20),
Times date
)
partitioned by (dates date)
row format delimited
fields terminated by ','
lines terminated by '\n'
stored as textfile;

sqoop import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--table add_test2 \
--where "cast(Times as date)='2021-06-29'" \
-m 1 \
--hive-import \
--hive-table sqooptest.sqp_par2 \
--hive-partition-key dates \
--hive-partition-value '2021-06-29' \
--fields-terminated-by ',' \
--lines-terminated-by '\n'

4.job

#创建job
sqoop job \
--create job_m2hdfs \
-- import \
--connect jdbc:mysql://wantest03:3306/sqooptest \
--username root \
--password 12345678 \
--query "select * from add_test2 where \$CONDITIONS" \
-m 1 \
--target-dir /sqooptest/jobm2hadd2 \
--fields-terminated-by '\t' \
--lines-terminated-by '\n' \
--check-column Times \
--incremental lastmodified \
--append \
--last-value '0000-00-00 00:00:00.0'

#查看job列表
sqoop job --list
#删除job
sqoop job --delete job_m2hdfs
#查看job定义
sqoop job --show job_m2hdfs
#执行job
sqoop job -exec job_m2hdfs