Sqoop看这篇文章就够了

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

Sqoop2 的最新版本是 1.99.7。请注意&＃xff0c;2 与 1 不兼容&＃xff0c;且特征不完整&＃xff0c;它并不打算用于
生产部署。
就是说Sqoop是一款用来在Hadoop&＃xff08;Hive&＃xff09;和关系型数据库之间传输数据的工具

Sqoop 在大数据体系位置&＃xff1a;

将导入或导出命令翻译成 mapreduce 程序来实现。在翻译出的 mapreduce 中主要是对inputformat 和 outputformat 进行定制。

官网&＃xff1a;http://sqoop.apache.org/
项目中一般用 Sqoop版本为&＃xff1a;sqoop-1.4.6

1. 安装 Sqoop 的前提是已经具备 Java 和Hadoop 的环境。 找到 对应版本安装包。
2. 解压指定路径&＃xff1a;

tar -zxf sqoop-1.4.6.bin__hadoop-2.0.4-alpha.tar.gz -C /opt/module/


3. 重命名配置文件

mv sqoop-env-template.sh sqoop-env.sh


4. 修改配置文件

export HADOOP_COMMON_HOME&＃61;/opt/module/hadoop-2.7.2
export HADOOP_MAPRED_HOME&＃61;/opt/module/hadoop-2.7.2
export HIVE_HOME&＃61;/opt/module/hive
export ZOOKEEPER_HOME&＃61;/opt/module/zookeeper-3.4.10
export ZOOCFGDIR&＃61;/opt/module/zookeeper-3.4.10
export HBASE_HOME&＃61;/opt/module/hbase


5. 拷贝JDBC驱动到Sqoop lib 下

cp mysql-connector-java-5.1.27.jar /opt/module/sqoop-1.4.6/lib/


6. 验证安装sqoop help

20/06/11 10:41:00 INFO sqoop.Sqoop: Running Sqoop version: 1.4.6
usage: sqoop COMMAND [ARGS]Available commands:codegen Generate code to interact with database recordscreate-hive-table Import a table definition into Hiveeval Evaluate a SQL statement and display the resultsexport Export an HDFS directory to a database tablehelp List available commandsimport Import a table from a database to HDFSimport-all-tables Import tables from a database to HDFSimport-mainframe Import datasets from a mainframe server to HDFSjob Work with saved jobslist-databases List available databases on a serverlist-tables List available tables in a databasemerge Merge results of incremental importsmetastore Run a standalone Sqoop metastoreversion Display version informationSee &＃39;sqoop help COMMAND&＃39; for information on a specific command.


7. 测试Sqoop是否可以连接数据库

sqoop list-databases --connect jdbc:mysql://10.100.22.94:9306/ --username user --password pwd
结果&＃xff1a;
information_schema
account
charbot
pkreg
hue
kg_ans

Sqoop 案例

导入数据&＃xff1a;在Sqoop中&＃xff0c;导入 概念指&＃xff1a;从非大数据集群&＃xff08;RDBMS&＃xff09;向大数据集群&＃xff08;HDFS&＃xff0c;HIVE&＃xff0c;HBASE&＃xff09;中传输数据&＃xff0c;使用import关键字。

RDBMS到HDFS

1.开启MySQL
2. 在MySQL中创建表格跟临时数据

mysql -uroot -p000000
mysql> create database company;
mysql> create table company.staff(id int(4) primary key not null auto_increment, name varchar(255), sex varchar(255));
mysql> insert into company.staff(name, sex) values(&＃39;Thomas&＃39;, &＃39;Male&＃39;);
mysql> insert into company.staff(name, sex) values(&＃39;Catalina&＃39;, &＃39;FeMale&＃39;);


导入数据到HDFS

1. 全部导入数据

bin/sqoop import \
--connect jdbc:mysql://IP地址:MySQL端口号/databaseName \
--username 账号 \
--password 密码 \
--table table名字 \
--target-dir /user/company \ (HDFS文件系统中的路径)
--delete-target-dir \ (先删除目标路径)
--num-mappers 1 \ (用几个mapper)
--fields-terminated-by "\t" (导出的数据列跟列用什么分割)


2. 查询导入

通过mysql中的query查询语句选择性导入数据

$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--target-dir /user/company \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by "\t" \
--query &＃39;select name,sex from staff where id <&＃61;1 and $CONDITIONS;&＃39;


提示&＃xff1a;must contain &＃39;$CONDITIONS&＃39; in WHERE clause.
提示&＃xff1a;如果query后使用的是双引号&＃xff0c;则$CONDITIONS前必须加转移符&＃xff0c;$CONDITIONS 防止shell识别为自己的变量。

$CONDITIONS 意义

1. 只要有–query&＃43;sql&＃xff0c;就需要加$CONDITIONS&＃xff0c;哪怕只有一个maptask。

sqoop import --connect jdbc:mysql://node01:3306/sqooptest --username root --password 123456 \
--target-dir /sqoop/conditiontest --delete-target-dir \
--query ‘select * from Person where score>50‘ --m 1


直接报错

ERROR ...must contain ‘$CONDITIONS‘ in WHERE clause.


2. 如果只有一个maptask&＃xff0c;可以不加–split-by来区分数据&＃xff0c;因为处理的是整份数据&＃xff0c;无需切分。

sqoop import --connect jdbc:mysql://node01:3306/sqooptest --username root --password 123456 \
--target-dir /sqoop/conditiontest --delete-target-dir \
--query ‘select * from Person where score>50 and $CONDITIONS‘ --m 1


日志

select * from Person where score>50 and (1 &＃61; 0)


处理后结果&＃xff1a;

3. 如果只有多个maptask&＃xff0c;需使用--split-by来区分数据&＃xff0c;$CONDITIONS替换查询范围。

sqoop import --connect jdbc:mysql://node01:3306/sqooptest --username root --password 123456 \
--target-dir /sqoop/conditiontest --delete-target-dir \
--query ‘select * from Person where score>1 and $CONDITIONS‘ --m 2 --split-by id


中间日志&＃xff1a;

# 查询id范围&＃xff0c;最大id&＃xff0c;最小id
20/02/07 12:19:55 INFO db.DataDrivenDBInputFormat: BoundingValsQuery: SELECT MIN(id), MAX(id) FROM (select * from Person where score>1 and (1 &＃61; 1) ) AS t1
# id的范围是1-5,2个分片
20/02/07 12:19:55 INFO db.IntegerSplitter: Split size: 2; Num splits: 2 from: 1 to: 5
20/02/07 12:19:55 INFO mapreduce.JobSubmitter: number of splits:2


结果&＃xff1a;

原理解析&＃xff1a;

当sqoop使用--query&＃43;sql执行多个maptask并行运行导入数据时&＃xff0c;每个maptask将执行一部分数据的导入&＃xff0c;原始数据需要使用--split-by某个字段‘来切分数据&＃xff0c;不同的数据交给不同的maptask去处理。maptask执行sql副本时&＃xff0c;需要在where条件中添加$CONDITIONS条件&＃xff0c;这个是linux系统的变量&＃xff0c;可以根据sqoop对边界条件的判断&＃xff0c;来替换成不同的值&＃xff0c;这就是说若split-by id&＃xff0c;则sqoop会判断id的最小值和最大值判断id的整体区间&＃xff0c;然后根据maptask的个数来进行区间拆分&＃xff0c;每个maptask执行一定id区间范围的数值导入任务&＃xff0c;如下为示意图。简而言之 保证数据结果有序化。

3. 导入指定列

$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root --password 000000 \
--target-dir /user/company --delete-target-dir \
--num-mappers 1 --fields-terminated-by "\t" \
--columns id,sex --table staff------指定MySQL中若干列导入到hive中
sqoop import -D mapreduce.job.queuename&＃61;hive账户 \
--connect jdbc:mysql://IP:9306/kg_db --username mysql用户 --password mysql密码 \
--table dim_risk_refuse_code --hive-import \
--hive-table dm_kg.dim_risk_refuse_code \
--columns "level,status,mreason,mreason_cn,subreason,subreaon_cn" -m 1

提示&＃xff1a;columns中如果涉及到多列&＃xff0c;用逗号分隔&＃xff0c;分隔时不要添加空格。

4. 使用sqoop关键字筛选查询导入数据

$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--target-dir /user/company \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by "\t" \
--table staff \
--where "id&＃61;1"


RDBMS到Hive

$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--num-mappers 1 \
--hive-import \
--fields-terminated-by "\t" \
--hive-overwrite \
--hive-table HiveTableName


提示&＃xff1a;该过程分为两步&＃xff0c;第一步将数据导入到HDFS&＃xff0c;第二步将导入到HDFS的数据迁移到Hive仓库&＃xff0c;第一步默认的临时目录是/user/atguigu/表名

如果 Hive 中已经有table了&＃xff0c;此时将MySQL数据导入到hive中无非就是按照hive格式将数据从MySQL中导入到hiveTable对应的HDFS路径

sqoop export -D mapreduce.job.queuename&＃61;MR账号 --connect jdbc:mysql://MySQLIP:9306/dbName --username mysql账号 --password MySQL密码 \
-table MySQLtableName \
--export-dir "hdfs://cluster/user/hive/warehouse/dm_kg.db/dwd_intopieces/etl_date&＃61;2020-06-11"
--input-fields-terminated-by &＃39;\001&＃39; \
--lines-terminated-by &＃39;\n&＃39; \
--input-null-string &＃39;\\N&＃39; \
--input-null-non-string &＃39;\\N&＃39;


RDBMS到Hbase

$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table company \
--columns "id,name,sex" \
--column-family "info" \
--hbase-create-table \
--hbase-row-key "id" \
--hbase-table "hbase_company" \
--num-mappers 1 \
--split-by id


提示&＃xff1a;sqoop1.4.6只支持HBase1.0.1之前的版本的自动创建HBase表的功能
解决方案&＃xff1a;手动创建HBase表

hbase> create &＃39;hbase_company,&＃39;info&＃39;


在HBase中scan这张表得到如下内容

hbase> scan ‘hbase_company’


导出数据

在Sqoop中&＃xff0c;导出概念指&＃xff1a;从大数据集群&＃xff08;HDFS&＃xff0c;HIVE&＃xff0c;HBASE&＃xff09;向非大数据集群&＃xff08;RDBMS&＃xff09;中传输数据&＃xff0c;叫做&＃xff1a;导出&＃xff0c;即使用export关键字。
HIVE/HDFS到RDBMS

$ bin/sqoop export \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--num-mappers 1 \
--export-dir /user/hive/warehouse/staff_hive \
--input-fields-terminated-by "\t"


提示&＃xff1a;Mysql中如果表不存在&＃xff0c;不会自动创建。需要按照hive、HDFS 中数据格式 先提前创建好MySQLTable。

脚本打包

使用opt格式的文件打包sqoop命令&＃xff0c;然后执行

1. 创建一个.opt文件

$ touch job_HDFS2RDBMS.opt


2. 编写sqoop脚本

$ vi job_HDFS2RDBMS.opt export --connect jdbc:mysql://hadoop102:3306/company --username root --password 000000
--table staff
--num-mappers 1
--export-dir /user/hive/warehouse/staff_hive
--input-fields-terminated-by "\t"


3. 执行该脚本

$ bin/sqoop --options-file job_HDFS2RDBMS.opt

Sqoop一些常用命令及参数

这里给大家列出来了一部分Sqoop操作时的常用参数&＃xff0c;以供参考&＃xff0c;需要深入学习的可以参看对应类的源代码

命令&参数详解

刚才列举了一些Sqoop的常用命令&＃xff0c;对于不同的命令&＃xff0c;有不同的参数&＃xff0c;让我们来一一列举说明。
首先来我们来介绍一下公用的参数&＃xff0c;所谓公用参数&＃xff0c;就是大多数命令都支持的参数。

公用参数&＃xff1a;数据库连接

公用参数&＃xff1a;import

公用参数&＃xff1a;export

公用参数&＃xff1a;hive

sqoop import -D mapreduce.job.queuename&＃61;kg \
--connect jdbc:mysql://IP:9306/kg_db --username user --password pwd \
--table kggn_addr_parsed \
--hive-overwrite -m 1 \
--hive-table db.tableName \
--autoreset-to-one-mapper \
--hive-partition-key etl_date \
--hive-partition-value $load_date \
--hive-drop-import-delims \
--fields-terminated-by "\0001" \
--target-dir /user/hive/warehouse/dm_kg.db/tablename/etl_date&＃61;2020-06-11 \
--delete-target-dir \
--null-string &＃39;\\N&＃39; \
--null-non-string &＃39;\\N&＃39;


公用参数介绍完之后&＃xff0c;我们来按照命令介绍命令对应的特有参数。

命令&参数&＃xff1a;import

将关系型数据库中的数据导入到HDFS&＃xff08;包括Hive&＃xff0c;HBase&＃xff09;中&＃xff0c;如果导入的是Hive&＃xff0c;那么当Hive中没有对应表时&＃xff0c;则自动创建。

1. 命令&＃xff1a;
   如&＃xff1a;导入数据到hive中

$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--hive-import


如&＃xff1a;增量导入数据到hive中&＃xff0c;mode&＃61;append

append导入&＃xff1a;
$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--num-mappers 1 \
--fields-terminated-by "\t" \
--target-dir /user/hive/warehouse/staff_hive \
--check-column id \
--incremental append \
--last-value 3


尖叫提示&＃xff1a;append不能与–hive-等参数同时使用&＃xff08;Append mode for hive imports is not yet supported. Please remove the parameter --append-mode&＃xff09;

如&＃xff1a;增量导入数据到hdfs中&＃xff0c;mode&＃61;lastmodified

先在mysql中建表并插入几条数据&＃xff1a;
mysql> create table company.staff_timestamp(id int(4), name varchar(255), sex varchar(255), last_modified timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP);
mysql> insert into company.staff_timestamp (id, name, sex) values(1, &＃39;AAA&＃39;, &＃39;female&＃39;);
mysql> insert into company.staff_timestamp (id, name, sex) values(2, &＃39;BBB&＃39;, &＃39;female&＃39;);
先导入一部分数据&＃xff1a;
$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff_timestamp \
--delete-target-dir \
--m 1
再增量导入一部分数据&＃xff1a;
mysql> insert into company.staff_timestamp (id, name, sex) values(3, &＃39;CCC&＃39;, &＃39;female&＃39;);
$ bin/sqoop import \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff_timestamp \
--check-column last_modified \
--incremental lastmodified \
--last-value "2017-09-28 22:20:38" \
--m 1 \
--append


尖叫提示&＃xff1a;使用lastmodified方式导入数据要指定增量数据是要–append&＃xff08;追加&＃xff09;还是要–merge-key&＃xff08;合并&＃xff09;
尖叫提示&＃xff1a;last-value指定的值是会包含于增量导入的数据中

参数&＃xff1a;

命令&参数&＃xff1a;export

从HDFS&＃xff08;包括Hive和HBase&＃xff09;中奖数据导出到关系型数据库中。

1. 命令&＃xff1a;

$ bin/sqoop export \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--export-dir /user/company \
--input-fields-terminated-by "\t" \
--num-mappers 1


参数&＃xff1a;

命令&参数&＃xff1a;codegen

将关系型数据库中的表映射为一个Java类&＃xff0c;在该类中有各列对应的各个字段

$ bin/sqoop codegen \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--bindir /home/admin/Desktop/staff \
--class-name Staff \
--fields-terminated-by "\t"


命令&参数&＃xff1a;create-hive-table

生成与关系数据库表结构对应的hive表结构。
命令&＃xff1a;

$ bin/sqoop create-hive-table \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--hive-table hive_staff


命令&参数&＃xff1a;eval

可以快速的使用SQL语句对关系型数据库进行操作&＃xff0c;经常用于在import数据之前&＃xff0c;了解一下SQL语句是否正确&＃xff0c;数据是否正常&＃xff0c;并可以将结果显示在控制台。
命令&＃xff1a;

$ bin/sqoop eval \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--query "SELECT * FROM staff"


命令&参数&＃xff1a;import-all-tables

可以将RDBMS中的所有表导入到HDFS中&＃xff0c;每一个表都对应一个HDFS目录
命令&＃xff1a;

$ bin/sqoop import-all-tables \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--warehouse-dir /all_tables


命令&参数&＃xff1a;job

用来生成一个sqoop任务&＃xff0c;生成后不会立即执行&＃xff0c;需要手动执行。

$ bin/sqoop job \--create myjob -- import-all-tables \--connect jdbc:mysql://hadoop102:3306/company \--username root \--password 000000
$ bin/sqoop job \
--list
$ bin/sqoop job \
--exec myjob


尖叫提示&＃xff1a;注意import-all-tables和它左边的–之间有一个空格
尖叫提示&＃xff1a;如果需要连接metastore&＃xff0c;则–meta-connect jdbc:hsqldb:hsql://linux01:16000/sqoop

尖叫提示&＃xff1a;在执行一个job时&＃xff0c;如果需要手动输入数据库密码&＃xff0c;可以做如下优化

<property><name>sqoop.metastore.client.record.passwordname><value>truevalue><description>If true, allow saved passwords in the metastore.description>
property>


命令&参数&＃xff1a;list-databases

$ bin/sqoop list-databases \
--connect jdbc:mysql://hadoop102:3306/ \
--username root \
--password 000000


参数&＃xff1a;与公用参数一样

命令&参数&＃xff1a;list-tables

$ bin/sqoop list-tables \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000


命令&参数&＃xff1a;merge

将HDFS中不同目录下面的数据合并在一起并放入指定目录中
数据环境

new_staff
1 AAA male
2 BBB male
3 CCC male
4 DDD male
old_staff
1 AAA female
2 CCC female
3 BBB female
6 DDD female


尖叫提示&＃xff1a;上边数据的列之间的分隔符应该为\t&＃xff0c;行与行之间的分割符为\n&＃xff0c;如果直接复制&＃xff0c;请检查之。
命令&＃xff1a;
如&＃xff1a;

创建JavaBean&＃xff1a;
$ bin/sqoop codegen \
--connect jdbc:mysql://hadoop102:3306/company \
--username root \
--password 000000 \
--table staff \
--bindir /home/admin/Desktop/staff \
--class-name Staff \
--fields-terminated-by "\t"开始合并&＃xff1a;
$ bin/sqoop merge \
--new-data /test/new/ \
--onto /test/old/ \
--target-dir /test/merged \
--jar-file /home/admin/Desktop/staff/Staff.jar \
--class-name Staff \
--merge-key id
结果&＃xff1a;
1 AAA MALE
2 BBB MALE
3 CCC MALE
4 DDD MALE
6 DDD FEMALE


命令&参数&＃xff1a;metastore

记录了Sqoop job的元数据信息&＃xff0c;如果不启动该服务&＃xff0c;那么默认job元数据的存储目录为~/.sqoop&＃xff0c;可在sqoop-site.xml中修改。
命令&＃xff1a;
如&＃xff1a;启动sqoop的metastore服务

bin/sqoop metastore