Sqoop

添码座原创大约 6 分钟

什么是Sqoop？

Sqoop是一款开源的中间件，它的主要作用是在关系型数据库（RDBMS）（例如MySQL）和大数据系统（例如Hadoop）之间进行数据传输。

Sqoop有两个大版本，可以在官方下载站点中查看。

Sqoop 1，就是1.4.x系列版本。
Sqoop 2，就是1.99.x系列版本。

这两个版本的主要区别就在于Sqoop 1只是一个客户端工具，可以执行指定的脚本。而Sqoop 2就集成了更多的服务，而且自身也服务化了。

本机安装

因为只是简单的数据传输工具，所以Sqoop 1就够用了。

MySQL和Hadoop可以使用已有的环境。

Docker部署MySQL。
Docker部署Hadoop或者本机部署Hadoop。

Sqoop的安装非常简单，简答的3步就可以完成。

下载并解压Sqoop。

> cd /home/work
> wget https://archive.apache.org/dist/sqoop/1.4.7/sqoop-1.4.7.bin__hadoop-2.6.0.tar.gz
> tar zxvf sqoop-1.4.7.bin__hadoop-2.6.0.tar.gz
> mv sqoop-1.4.7.bin__hadoop-2.6.0 sqoop-1.4.7

配置环境变量文件。

> vi /etc/profile

# 添加以下内容
# SQOOP
export SQOOP_HOME=/home/work/sqoop-1.4.7
export PATH=.:$SQOOP_HOME/bin:$PATH

将MySQL驱动包（例如mysql-connector-j-8.0.33.jar）和commons-lang-2.6.jar这两个jar包拷贝到${SQOOP_HOME}/lib目录中。

数据导入

因为Sqoop属于大数据生态的一部分，所以它所说的导入显然是从传统的关系型数据库（RDBMS）导入到大数据系统，也就是将数据从MySQL导入到Hadoop。

Sqoop官方的导入文档描述的比较详细，可以直接通过文档实现数据的导入功能。

Sqoop的导入可以分为两种。

全表导入：直接将MySQL某个表中的全部内容导入到Hadoop，适合于系统环境初始化的场景。
查询导入：通过查询语句将MySQL某个表中的部分内容导入到Hadoop，适合于日常增量消费数据的场景。

先在MySQL中创建用于测试的数据库和表。

-- 创建数据库
mysql> CREATE DATABASE itechthink;
mysql> USE itechthink;

-- 创建数据表
mysql> DROP TABLE IF EXISTS t_import_user;
mysql> CREATE TABLE t_import_user (
         id int(11),
         name varchar(32)
       );
mysql> INSERT INTO t_import_user VALUES(1, "lixingyun"), (2, "wanglin"), (3, "xiaoyan"), (4, "luofeng");

全表导入

下面是一个全表导入的Sqoop脚本。

> sqoop import \
    --connect jdbc:mysql://172.16.185.176:3306/itechthink?serverTimezone=UTC \
    --username root \
    --password 123456 \
    --table t_import_user \
    --target-dir /sqoop/out1 \
    # 删除之前导出的内容
    --delete-target-dir \
    # 如果表中没有主键且没有`--num-mappers`参数，那么将执行失败，该参数如果不指定的话默认值为4
    --num-mappers 1 \
    --fields-terminated-by '\t'

脚本在命令行中的执行结果如下。

......

INFO sqoop.Sqoop: Running Sqoop version: 1.4.7

......

INFO mapreduce.Job:  map 0% reduce 0%
INFO mapreduce.Job:  map 100% reduce 0%
INFO mapreduce.Job: Job job_1681562327599_0001 completed successfully
INFO mapreduce.Job: Counters: 33
File System Counters
	FILE: Number of bytes read=0
	FILE: Number of bytes written=230024
	FILE: Number of read operations=0
	FILE: Number of large read operations=0
	FILE: Number of write operations=0
	HDFS: Number of bytes read=87
	HDFS: Number of bytes written=42
	HDFS: Number of read operations=6
	HDFS: Number of large read operations=0
	HDFS: Number of write operations=2
	HDFS: Number of bytes read erasure-coded=0
Job Counters 
	Launched map tasks=1
	Other local map tasks=1
	Total time spent by all maps in occupied slots (ms)=23269
	Total time spent by all reduces in occupied slots (ms)=0
	Total time spent by all map tasks (ms)=23269
	Total vcore-milliseconds taken by all map tasks=23269
	Total megabyte-milliseconds taken by all map tasks=23827456
Map-Reduce Framework
	Map input records=4
	Map output records=4
	Input split bytes=87
	Spilled Records=0
	Failed Shuffles=0
	Merged Map outputs=0
	GC time elapsed (ms)=113
	CPU time spent (ms)=1720
	Physical memory (bytes) snapshot=241238016
	Virtual memory (bytes) snapshot=2594168832
	Total committed heap usage (bytes)=246415360
	Peak Map Physical memory (bytes)=241238016
	Peak Map Virtual memory (bytes)=2594168832
File Input Format Counters 
	Bytes Read=0
File Output Format Counters 
	Bytes Written=42
INFO mapreduce.ImportJobBase: Transferred 42 bytes in 174.3538 seconds (0.2409 bytes/sec)
INFO mapreduce.ImportJobBase: Retrieved 4 records.

在Hadoop中查看数据导入的结果。

> hdfs dfs -cat /sqoop/out1/*
1    lixingyun
2    wanglin
3    xiaoyan
4    luofeng

查询导入

下面是一个查询导入的Sqoop脚本。

> sqoop import \
    --connect jdbc:mysql://172.16.185.176:3306/itechthink?serverTimezone=UTC \
    --username root \
    --password 123456 \
    --target-dir /sqoop/out2 \
    --delete-target-dir \
    --num-mappers 1 \
    --fields-terminated-by '\t' \
    # 该参数和`--table`不能同时出现
    # `$CONDITIONS`属于一种较为固定形式的写法，没啥意义，但必须要加
    --query 'SELECT id, name FROM t_import_user WHERE id < 3 AND $CONDITIONS'

执行完成后可以在Hadoop中查看数据导入的结果。

> hdfs dfs -cat /sqoop/out2/*
1    lixingyun
2    wanglin

如果数据表中存在NULL值，Sqoop会直接输出null字符串。

例如，在t_import_user表中新插入两条数据，分别让id和name都出现NULL值，执行导入后结果如下。

> hdfs dfs -cat /sqoop/out1/*
1       lixingyun
2       wanglin
3       xiaoyan
4       luofeng
null    tangsan
5       null

针对这种情况有两种解决方法。

不允许MySQL表的字段为空（NULL），或者给字段设置一个默认值。
给Sqoop增加导入参数。

> sqoop import \
    --connect jdbc:mysql://172.16.185.176:3306/itechthink?serverTimezone=UTC \
    --username root \
    --password 123456 \
    --table t_import_user \
    --target-dir /sqoop/out1 \
    --delete-target-dir \
    --num-mappers 1 \
    --fields-terminated-by '\t' \
    # 字符串字段为空时用`default`替换
    --null-string 'default' \
    # 数字型字段为空时用`-1`替换
    --null-non-string '-1'

执行完成后在Hadoop中查看数据为NULL时导入的替换结果。

> hdfs dfs -cat /sqoop/out1/*
1   lixingyun
2   wanglin
3   xiaoyan
4   luofeng
-1  tangsan
5   default

Hive表中的NULL用\N表示，也可以将null字符串替换为\\N，方便处理。

> sqoop import \
    ......
    --null-string '\\N' \
    --null-non-string '\\N'

替换完成后在Hadoop中的结果显示如下。

> hdfs dfs -cat /sqoop/out1/*
1   lixingyun
2   wanglin
3   xiaoyan
4   luofeng
\N  tangsan
5   \N

数据导出

这是Sqoop官方的导出文档。

导出是和导入相反的过程，将Hadoop中的数据导出到MySQL。

先在MySQL中创建用于测试的数据表。

-- 创建数据表
mysql> DROP TABLE IF EXISTS t_export_user;
mysql> CREATE TABLE t_export_user (
         id int(11),
         name varchar(32)
       );

下面是Sqoop的导出脚本。

> sqoop export \
    --connect jdbc:mysql://172.16.185.176:3306/itechthink?serverTimezone=UTC \
    --username root \
    --password 123456 \
    --table t_export_user \
    --export-dir /sqoop/out1 \
    --input-fields-terminated-by '\t'

导入完成后在MySQL中查询结果如下。

mysql> SELECT * FROM t_export_user;
+------+-----------+
| id   | name      |
+------+-----------+
|    1 | lixingyun |
|    3 | xiaoyan   |
|    2 | wanglin   |
|    4 | luofeng   |
+------+-----------+

如果导入的Hadoop数据中存在null字符串，导入后会是什么？

Sqoop会将null字符串原样地还原为MySQL中的NULL。

mysql> SELECT * FROM t_export_user;
+------+-----------+
| id   | name      |
+------+-----------+
|    4 | luofeng   |
| NULL | tangsan   |
|    1 | lixingyun |
|    2 | wanglin   |
|    3 | xiaoyan   |
|    5 | NULL      |
+------+-----------+

另外，导入时还可以通过指定选项，实现INSERT和UPDATE的功能。

如果要实现这些功能，那么表里面必须要有主键字段。

给表t_export_user指定主键字段。

mysql> DROP TABLE IF EXISTS t_export_user;
mysql> CREATE TABLE t_export_user (
         id int(11) primary key,
         name varchar(32)
       );

-- 插入两条数据
mysql> INSERT INTO t_export_user VALUES(1, "xiaoyan");
mysql> SELECT * FROM t_export_user;
+----+---------+
| id | name    |
+----+---------+
|  1 | xiaoyan |
+----+---------+

之前在查询导入时，Hadoop里存在如下数据。

> hdfs dfs -cat /sqoop/out2/*
1    lixingyun
2    wanglin

通过下面的脚本实现存在则更新，不存在则新增的功能。

> sqoop export \
    --connect jdbc:mysql://172.16.185.176:3306/itechthink?serverTimezone=UTC \
    --username root \
    --password 123456 \
    --table t_export_user \
    --export-dir /sqoop/out2 \
    --input-fields-terminated-by '\t' \
    # 存在则更新，不存在则新增
    --update-key id \
    --update-mode allowinsert

执行导出后MySQL中的数据如下。

mysql> SELECT * FROM t_export_user;
+------+-----------+
| id   | name      |
+------+-----------+
|    1 | lixingyun |
|    2 | wanglin   |
+------+-----------+

感谢支持

更多内容，请移步《超级个体》。