Sqoop全量数据导入、增量数据导入、并发导入（Sqoop进阶）

曾经终败给现在 2024-04-17 20:42 10阅读 0赞

### 一、sqoop概述： ###

Sqoop(发音：skup)是一款开源的工具，主要用于在Hadoop(Hive)与传统的数据库(mysql、postgresql...)间进行数据的传递，可以将一个关系型数据库（例如 ： MySQL ,Oracle ,Postgres等）中的数据导进到Hadoop的HDFS中，也可以将HDFS的数据导进到关系型数据库中。

Sqoop项目开始于2009年，最早是作为Hadoop的一个第三方模块存在，后来为了让使用者能够快速部署，也为了让开发人员能够更快速的迭代开发，Sqoop独立成为一个[Apache][]项目。

**       Sqoop是一个在结构化数据和Hadoop之间进行批量数据迁移的工具，结构化数据可以是Mysql、Oracle等RDBMS。Sqoop底层用MapReduce程序实现抽取、转换、加载，MapReduce天生的特性保证了并行化和高容错率，而且相比Kettle等传统ETL工具，任务跑在Hadoop集群上，减少了ETL服务器资源的使用情况。在特定场景下，抽取过程会有很大的性能提升。**

如果要用Sqoop，必须正确安装并配置Hadoop，因依赖于本地的hadoop环境启动MR程序；mysql、oracle等数据库的JDBC驱动也要放到Sqoop的lib目录下

**import**

import是数据从RDBMS导入到Hadoop的工具

**export**

export是HDFS里的文件导出到RDBMS的工具，不能从hive、hbase导出数据，且HDFS文件只能是文本格式。如果要把hive表数据导出到RDBMS，可以先把hive表通过查询写入到一个临时表，临时用文本格式，然后再从该临时表目录里export数据。

Sqoop从HDFS目录里读取文件，所以启动的map task数依赖于-m参数、文件大小、文件数量、块大小等，[可以参考这篇文章][Link 1]

参考：[https://www.jianshu.com/p/be33f4b5c62e][https_www.jianshu.com_p_be33f4b5c62e]

**二、Sqoop支持两种方式的全量数据导入和增量数据导入，同时可以指定数据是否以并发形式导入。**

下面依次来看：

**（1）全量数据导入**  
   就像名字起的那样，全量数据导入就是一次性将所有需要导入的数据，从关系型数据库一次性地导入到Hadoop中（可以是HDFS、Hive等）。全量导入形式使用场景为一次性离线分析场景。用sqoop import命令，具体如下：

> \# 全量数据导入  
> sqoop import \\  
>  --connect jdbc:mysql://192.168.xxx.xxx:3316/testdb \\  
>  --username root \\  
>  --password 123456 \\  
>  --query “select \* from test\_table where \\$CONDITIONS” \\  
>  --target-dir /user/root/person\_all \\   
>  --fields-terminated-by “,” \\  
>  --hive-drop-import-delims \\  
>  --null-string “\\\\N” \\  
>  --null-non-string “\\\\N” \\  
>  --split-by id \\  
>  -m 6 \\

重要参数说明：

参数    说明  
– query    SQL查询语句  
– target-dir    HDFS目标目录（确保目录不存在，否则会报错，因为Sqoop在导入数据至HDFS时会自己在HDFS上创建目录）  
–hive-drop-import- delims    删除数据中包含的Hive默认分隔符（^A, ^B, \\n）  
–null-string    string类型空值的替换符（Hive中Null用\\n表示）  
–null-non-string    非string类型空值的替换符  
–split-by    数据切片字段（int类型，m>1时必须指定）  
\-m    Mapper任务数，默认为4

**（2）增量数据导入**  
事实上，在生产环境中，系统可能会定期从与业务相关的关系型数据库向Hadoop导入数据，导入数仓后进行后续离线分析。故我们此时不可能再将所有数据重新导一遍，此时我们就需要增量数据导入这一模式了。

增量数据导入分两种，一是基于递增列的增量数据导入（Append方式）。二是基于时间列的增量数据导入（LastModified方式）。

**    1、Append方式**

举个栗子，有一个订单表，里面每个订单有一个唯一标识自增列ID，在关系型数据库中以主键形式存在。之前已经将id在0~5201314之间的编号的订单导入到Hadoop中了（这里为HDFS），现在一段时间后我们需要将近期产生的新的订单数据导入Hadoop中（这里为HDFS），以供后续数仓进行分析。此时我们只需要指定–incremental 参数为append，–last-value参数为5201314即可。表示只从id大于5201314后开始导入。

> \# Append方式的全量数据导入  
>  sqoop import \\  
>    --connect jdbc:mysql://192.168.xxx.xxx:3316/testdb \\  
>    --username root \\  
>    --password 123456 \\  
>    --query “select order\_id, name from order\_table where \\$CONDITIONS” \\  
>    --target-dir /user/root/orders\_all \\   
>    --split-by order\_id \\  
>    -m 6  \\  
>    --incremental append \\  
>    --check-column order\_id \\  
>    --last-value 5201314

重要参数说明：

参数    说明  
–incremental append    基于递增列的增量导入（将递增列值大于阈值的所有数据增量导入Hadoop）  
–check-column    递增列（int）  
–last-value    阈值（int）

**2、lastModify方式 **  
此方式要求原有表中有time字段，它能指定一个时间戳，让Sqoop把该时间戳之后的数据导入至Hadoop（这里为HDFS）。因为后续订单可能状态会变化，变化后time字段时间戳也会变化，此时Sqoop依然会将相同状态更改后的订单导入HDFS，当然我们可以指定merge-key参数为orser\_id，表示将后续新的记录与原有记录合并。  
\# 将时间列大于等于阈值的数据增量导入HDFS

>  sqoop import \\  
>    --connect jdbc:mysql://192.168.xxx.xxx:3316/testdb \\  
>    --username root \\  
>    --password transwarp \\  
>    --query “select order\_id, name from order\_table where \\$CONDITIONS” \\  
>    --target-dir /user/root/order\_all \\   
>    --split-by id \\  
>    -m 4  \\  
>    --incremental lastmodified \\  
>    --merge-key order\_id \\  
>    --check-column time \\  
>    \# remember this date !!!  
>    --last-value “2014-11-09 21:00:00”

重要参数说明：

参数    说明  
–incremental lastmodified    基于时间列的增量导入（将时间列大于等于阈值的所有数据增量导入Hadoop）  
–check-column    时间列（int）  
–last-value    阈值（int）  
–merge-key    合并列（主键，合并键值相同的记录）  
**（3）并发导入参数如何设置？**  
我们知道通过 -m 参数能够设置导入数据的 map 任务数量，即指定了 -m 即表示导入方式为并发导入，这时我们必须同时指定 - -split-by 参数指定根据哪一列来实现哈希分片，从而将不同分片的数据分发到不同 map 任务上去跑，避免数据倾斜。

**重要Tip：**

> 生产环境中，为了防止主库被Sqoop抽崩，我们一般从备库中抽取数据。  
> 一般RDBMS的导出速度控制在60~80MB/s，每个 map 任务的处理速度5~10MB/s 估算，即 -m 参数一般设置4~8，表示启动 4~8 个map 任务并发抽取。

**（4） 执行sqoop job**

sqoop job --exec fdc\_equipment\_job

如果创建的job有问题，可以删除再重新创建：

sqoop job --delete fdc\_equipment\_job

参考：https://blog.csdn.net/whdxjbw/article/details/81079746

[Apache]: https://baike.baidu.com/item/Apache/6265
[Link 1]: https://link.jianshu.com/?t=http%3A%2F%2Fmp.weixin.qq.com%2Fs%3F__biz%3DMzI0MzYzNTk5NQ%3D%3D%26mid%3D2247483821%26idx%3D1%26sn%3D47e21596f934814c148f00477eb2751c%26chksm%3De96b43d9de1ccacfe525e1d05bb8ec462a1d17eeb77ef8f8a915a824912fa091f129d7cadb7f%26scene%3D21%23wechat_redirect
[https_www.jianshu.com_p_be33f4b5c62e]: https://www.jianshu.com/p/be33f4b5c62e