11 辅助系统框架之sqoop数据迁移

976-沈同学

发表文章数:65

热门标签

首页 » 大数据 » 正文

11 辅助系统框架之sqoop数据迁移

1.概述

sqoop是apache旗下一款“Hadoop和关系数据库服务器之间传送数据”的工具。
导入数据:MySQL,Oracle导入数据到Hadoop的HDFS、HIVE、HBASE等数据存储系统;
导出数据:从Hadoop的文件系统中导出数据到关系数据库mysql等

通过MR的inputformat和outputformat来实现数据的输入与输出,底层执行的全部都是MR的任务,
只不过这个mr只有map阶段,没有reduce阶段
说白了只是对数据进行抽取,从一个地方,抽取到另外一个地方

2.sqoop1与sqoop2架构对比

一般用sqoop1.x比较多,直接将我们的命令写入到脚本里面去,执行脚本即可
sqoop 1.x 不用安装,解压就能用
11 辅助系统框架之sqoop数据迁移
sqoop 2.x 架构发生了变化,引入了一个服务端 可以通过代码提交sqoop的任务
11 辅助系统框架之sqoop数据迁移

3.工作机制

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

4.sqoop实战及原理

4.1 sqoop安装

安装sqoop的前提是已经具备java和hadoop的环境

1、下载并解压

下载地址

http://archive.cloudera.com/cdh5/cdh/5/
sqoop1版本详细下载地址
http://archive.cloudera.com/cdh5/cdh/5/sqoop-1.4.6-cdh5.14.0.tar.gz
sqoop2版本详细下载地址
http://archive.cloudera.com/cdh5/cdh/5/sqoop2-1.99.5-cdh5.14.0.tar.gz

我们这里使用sqoop1的版本,下载之后上传到/export/softwares目录下,然后进行解压

cd /export/softwares
tar -zxvf sqoop-1.4.6-cdh5.14.0.tar.gz -C ../servers/

2、修改配置文件

cd /export/servers/sqoop-1.4.6-cdh5.14.0/conf/
cp sqoop-env-template.sh  sqoop-env.sh
vim sqoop-env.sh

export HADOOP_COMMON_HOME=/export/servers/hadoop-2.6.0-cdh5.14.0
export HADOOP_MAPRED_HOME=/export/servers/hadoop-2.6.0-cdh5.14.0
export HIVE_HOME=/export/servers/hive-1.1.0-cdh5.14.0

3、加入额外的依赖包

sqoop的使用需要添加两个额外的依赖包

一个是mysql的驱动包,
另一个是java-json的的依赖包,不然就会报错
mysql-connector-java-5.1.40.jar
java-json.jar

将这个两个jar包添加到sqoop的lib目录下

4、验证启动

cd /export/servers/sqoop-1.4.6-cdh5.14.0
bin/sqoop-version

5.Sqoop的数据导入

“导入工具”导入单个表从RDBMS到HDFS。表中的每一行被视为HDFS的记录。所有记录都存储为文本文件的文本数据(或者Avro、sequence文件等二进制数据)

列举出所有的数据库与表

命令行查看帮助

bin/sqoop list-databases --help

列出win7主机所有的数据库

bin/sqoop list-databases --connect jdbc:mysql://node3:3306/ --username 	root --password 123456

查看某一个数据库下面的所有数据表

bin/sqoop list-tables --connect jdbc:mysql://node3:3306/hive --username root --password 123456

其它导入示例

1)表数据准备

在mysql中有一个库userdb中三个表:emp, emp_add和emp_conn
表emp:

source /export/servers/userdb.sql;

2)导入数据库表数据到HDFS

下面的命令用于从MySQL数据库服务器中的emp表导入HDFS。

bin/sqoop import --connect jdbc:mysql://node3:3306/userdb --password 123456 --username root --table emp -m 1
如果成功执行,那么数据默认会存储再/user/root/目录下
hdfs  dfs  -ls  /user/root/emp

3)导入到HDFS指定目录

在导入表数据到HDFS使用Sqoop导入工具,我们可以指定目标目录。

使用参数 --target-dir来指定导出目的地,
使用参数—delete-target-dir来判断导出目录是否存在,如果存在就删掉
bin/sqoop import  --connect jdbc:mysql://node3:3306/userdb --username root --password 123456 --delete-target-dir --table emp  --target-dir /sqoop/emp -m 1
hdfs dfs -text /sqoop/emp/part-m-00000 (查看导出的数据)

它会用逗号(,)分隔emp表的数据和字段。
11 辅助系统框架之sqoop数据迁移

4)导入到hdfs指定目录并指定字段之间的分隔符

bin/sqoop import  --connect jdbc:mysql://node3:3306/userdb --username root /
--password 123456 --delete-target-dir --table emp  --target-dir /sqoop/emp2 --m 1/
--fields-terminated-by '/t'

查看文件内容

hdfs dfs -text /sqoop/emp2/part-m-00000

11 辅助系统框架之sqoop数据迁移

5)导入关系表到HIVE

第一步:拷贝jar包

将我们mysql表当中的数据直接导入到hive表中的话,我们需要将hive的一个叫做hive-exec-1.1.0-cdh5.14.0.jar的jar包拷贝到sqoop的lib目录下

cp /export/servers/hive-1.1.0-cdh5.14.0/lib/hive-exec-1.1.0-cdh5.14.0.jar /export/servers/sqoop-1.4.6-cdh5.14.0/lib/
第二步:准备hive数据库与表

将我们mysql当中的数据导入到hive表当中来

create database sqooptohive;
use sqooptohive;
create external table emp_hive(id int,name string,deg string,salary int ,dept string) row format delimited fields terminated by '/001';
第三步:开始导入
bin/sqoop import --connect jdbc:mysql://node3:3306/userdb --username root /
--password 123456 --table emp --fields-terminated-by '/001' /
--hive-import --hive-table sqooptohive.emp_hive --hive-overwrite --delete-target-dir --m 1
第四步:hive表数据查看
select * from emp_hive;

11 辅助系统框架之sqoop数据迁移

6)导入关系表到hive并自动创建hive表

我们也可以通过命令来将我们的mysql的表直接导入到hive表当中去

bin/sqoop import --connect jdbc:mysql://node3:3306/userdb --username root /
--password 123456 --table emp_conn --hive-import -m 1 /
--hive-database sqooptohive;

通过这个命令,我们可以直接将我们mysql表当中的数据以及表结构一起倒入到hive当中去

7)导入表数据子集

我们可以导入表的使用Sqoop导入工具,"where"子句的一个子集。它执行在各自的数据库服务器相应的SQL查询,并将结果存储在HDFS的目标目录。
where子句的语法如下。

--where <condition>
按照条件进行查找,通过—where参数来查找表emp_add当中city字段的值为sec-bad的所有数据导入到hdfs上面去

bin/sqoop import /
--connect jdbc:mysql://node3:3306/userdb /
--username root --password 123456 --table emp_add /
--target-dir /sqoop/emp_add -m 1  --delete-target-dir /
--where "city = 'sec-bad'"

8)sql语句查找导入hdfs

我们还可以通过 –query参数来指定我们的sql语句,通过sql语句来过滤我们的数据进行导入

*使用sql语句来进行查找是不能加参数–table
*并且必须要添加where条件,
*并且where条件后面必须带一个$CONDITIONS 这个字符串,
*并且这个sql语句必须用单引号,不能用双引号

bin/sqoop import /
--connect jdbc:mysql://node3:3306/userdb --username root --password 123456 /
--delete-target-dir -m 1 /
--query 'select phno from emp_conn where 1=1 and  $CONDITIONS' /
--target-dir /sqoop/emp_conn

查看hdfs数据内容

hdfs dfs -text /sqoop/emp_conn/part*

9)增量导入

只导入我们部分需要的数据
现在时间2020-11-20 02:30:00 导入数据时间 2020-11-19 00:00:00 2018-11-19 23:59:59

全量导入,数据太多,对数据库压力比较大
增量导入是仅导入新添加的表中的行的技术。
它需要添加‘incremental’, ‘check-column’, 和 ‘last-value’选项来执行增量导入。
下面的语法用于Sqoop导入命令增量选项。
--incremental <mode>
--check-column <column name>
--last value <last check column value>

导入emp表当中id大于1202的所有数据
注意:增量导入的时候,一定不能加参数–delete-target-dir否则会报错

bin/sqoop import /
--connect jdbc:mysql://node3:3306/userdb /
--username root /
--password 123456 /
--table emp /
--incremental append /
--check-column id /
--last-value 1202  /
-m 1 /
--target-dir /sqoop/increment

查看数据内容(大于1202的被记录)

hdfs dfs -text /sqoop/increment/part*

11 辅助系统框架之sqoop数据迁移
因为每次不可能知道记录到了哪个id,如何解决增量导入的问题??

一般情况下数据仓库都会有以下字段
每个数据都会有一个创建时间,可以根据我们的创建时间来判断是否是我们前一天的数据

每个表都会有三个固定的字段
create_time 
update_time
is_delete
operator

如何解决导入减量数据的问题???

什么是减量数据????删除掉的数据  数据不是做真删除  
做假删除,其实就是改变了一些数据的状态,数据的更细时间,同步改变
银行客户  13859687451
变更手机号  13896541235

所有的减量数据都转化为变更数据来处理
第一个:涉及到数据的变更问题,
变更数据一定有更新时间   每天导入数据的时候,需要根据创建时间和更新时间来一起判断
第一条数据 create_time  2018-11-19 12:23:45 
第二条数据 update_time  2018-11-19 15:23:45
根据两个条件来同时进行判断,满足任意一个,都要将数据导入过来

id   create_time             update_time
1    2018-11-15 23:45:15     2018-11-15 23:45:15
1    2018-11-15 23:45:15     2018-11-28 23:45:15
group by  id 来进行操作

都是根据create_time  update_time 来联合进行判断

第二种增量导入通过–where条件来实现
或者我们使用–where来进行控制数据的选取会更加精准

bin/sqoop import /
--connect jdbc:mysql://node3:3306/userdb /
--username root /
--password 123456  /
--table emp /
--incremental append  /
--where "create_time > '2018-06-17 00:00:00' and is_delete='1' and create_time < '2018-06-17 23:59:59'" /
--target-dir /sqoop/incement2 /
--check-column id  /
--m 1

–last-value 2018-06-17 00:00:00

2018-06-18 01:00:00  产生一条数据
2018-06-18 02:00:00   这个不能完成这个任务

sqoop增量与减量导入:

增量数据导入:
每天导入前一天数据 一般都是使用 create_time update_time
增量导入通过记录上一次导入数据的id值,来确定下一次导入的数据

bin/sqoop import /
	--connect jdbc:mysql://192.168.22.22:3306/userdb /
	--username root /
	--password admin /
	--table emp /
	--incremental append /
	--check-column id /
	--last-value 1202  /
	-m 1 /
	--target-dir /sqoop/increment

使用id来做增量导入数据:第一个问题就是导入数据不准确,数据最后一个id没法记录
如果要使用增量导入,最好的方式卡时间点
create_time 2019-07-25 00:00:00 2019-07-25 23:59:59

bin/sqoop import /
--connect jdbc:mysql://192.168.22.22:3306/userdb /
--username root /
--password admin  /
--table emp /
--incremental append  /
--where "create_time > '2018-06-17 00:00:00'  and create_time < '2018-06-17 23:59:59'" /
--target-dir /sqoop/incement2 /
--check-column id  /
--m 1

建表的时候:
create_time默认值是当前的创建时间
update_time最开始默认值也是当前创建时间,一旦数据更新,就会变为更新的

//MYSQL5.6新特性
`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

减量数据导入:

减量数据怎么解决:删除的数据,假删除,更新操作 需要将更新的数据依据update_time导入过来
一条数据存在多个状态,做拉链表

有一个人去银行办业务,业务系统只有一条数据
第一天 手机号  2019-07-25 13598745612
第二天 手机号  2019-07-26 13598745613
第二天 手机号  2019-07-27 13598745614


业务系统当中,更新语句就行了,数据仓库当中怎么办????

数据仓库当中:(有更新的历史数据:也叫拉链表)

第一天 手机号  2019-07-25 13598745612
第二天 手机号  2019-07-26 13598745613
第二天 手机号  2019-07-27 13598745614

拉链表:依据时间的顺序,给数据进行排序

6.Sqoop的数据导出

hdfs导出到mysql

数据是在HDFS当中的如下目录/sqoop/emp,数据内容如下

1201,gopal,manager,50000,TP,2018-06-17 18:54:32.0,2018-06-17 18:54:32.0,1
1202,manisha,Proof reader,50000,TP,2018-06-15 18:54:32.0,2018-06-17 20:26:08.0,1
1203,khalil,php dev,30000,AC,2018-06-17 18:54:32.0,2018-06-17 18:54:32.0,1
1204,prasanth,php dev,30000,AC,2018-06-17 18:54:32.0,2018-06-17 21:05:52.0,0
1205,kranthi,admin,20000,TP,2018-06-17 18:54:32.0,2018-06-17 18:54:32.0,1

第一步:创建mysql表

truncate table emp_out; (已经有表清空表内数据)


CREATE TABLE `emp_out` (
  `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,
  `create_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `update_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `is_delete` BIGINT(20) DEFAULT '1'
) ENGINE=INNODB DEFAULT CHARSET=utf8;

第二步:执行导出命令

通过export来实现数据的导出,将hdfs的数据导出到mysql当中去

bin/sqoop export /
--connect jdbc:mysql://node3:3306/userdb /
--username root --password 123456 /
--table emp_out /
--export-dir /sqoop/emp /
--input-fields-terminated-by ","

第三步:验证mysql表数据

11 辅助系统框架之sqoop数据迁移
导出数据的时候,如果导出到一半,报错了怎么办????
一般都是创建mysql的临时表 如果临时表导入成功,再往目的表里面导入

sqoop的job

就是将我们的导入导出到命令,保存起来,下次可以直接调用,没有必要,写脚本就好了

7.java执行shell命令

需求描述:在实际工作中,总会有些时候需要我们通过java代码通过远程连接去linux服务器上面执行一些shell命令,包括一些集群的状态管理,执行任务,集群的可视化界面操作等等,所以我们可以通过java代码来执行linux服务器的shell命令
为了解决上述问题,google公司给提出了对应的解决方案,开源出来了一个jar包叫做sshxcute,通过这个jar包我们可以通过java代码,非常便捷的操作我们的linux服务器了

项目地址如下:
https://code.google.com/archive/p/sshxcute/
使用说明
https://www.ibm.com/developerworks/cn/opensource/os-sshxcute/

第一步:创建maven的java工程并导入jar包
11 辅助系统框架之sqoop数据迁移
第二步:开发测试用例

package cn.itcast.sshxcute;

import net.neoremind.sshxcute.core.ConnBean;
import net.neoremind.sshxcute.core.SSHExec;
import net.neoremind.sshxcute.exception.TaskExecFailException;
import net.neoremind.sshxcute.task.impl.ExecCommand;

public class XcuteExcute {
    public static void main(String[] args) throws TaskExecFailException {
        //获取SSHExec用于执行我们的shell命令
        SSHExec sshExec = SSHExec.getInstance(new ConnBean("node3", "root", "123456"));
        sshExec.connect();
        //执行命令
        //CustomTask抽象类,要么找子类,要么找这个类有没有方法返回实例
        ExecCommand execCommand = new ExecCommand("echo 1 >> /export/servers/helloworld.txt");
        sshExec.exec(execCommand);
        sshExec.disconnect();
    }
}

总结

sqoop导入导出是一个离线处理的工具
底层使用的都是MR的程序
有没有什么工具,可以实现实时的抽取数据
canal 通过解析binlog可以实现实时的数据抽取
flume 自定义source 代码在github上面 也可以实现近似实时的数据抽取
streamSet 比较强大,可以实现实时的抽取数据
11 辅助系统框架之sqoop数据迁移

标签:

拜师教育学员文章:作者:976-沈同学, 转载或复制请以 超链接形式 并注明出处 拜师资源博客
原文地址:《11 辅助系统框架之sqoop数据迁移》 发布于2020-04-24

分享到:
赞(0) 打赏

评论 抢沙发

评论前必须登录!

  注册



长按图片转发给朋友

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

Vieu3.3主题
专业打造轻量级个人企业风格博客主题!专注于前端开发,全站响应式布局自适应模板。

登录

忘记密码 ?

您也可以使用第三方帐号快捷登录

Q Q 登 录
微 博 登 录