Sqoop1.4.4 实现将 Oracle10g 中的增量数据导入

女爷i 2022-08-05 07:29 130阅读 0赞

Sqoop1.4.4 实现将 Oracle10g 中的增量数据导入 Hive0.13.1 ，并更新Hive中的主表  
  
  
需求  
将Oracle中的业务基础表增量数据导入Hive中，与当前的全量表合并为最新的全量表。  
设计  
涉及的三张表：  
全量表：保存了截止上一次同步时间的全量基础数据表  
增量表：增量临时表  
更新后的全量表：更新后的全量数据表  
   
步骤：  
通过Sqoop将Oracle中的表导入Hive，模拟全量表和增量表  
通过Hive将“全量表+增量表”合并为“更新后的全量表”，覆盖当前的全量表  
  
  
步骤1：通过Sqoop将Oracle中表的导入Hive，模拟全量表和增量表  
为了模拟场景，需要一张全量表，和一张增量表，由于数据源有限，所以两个表都来自Oracle中的OMP\_SERVICE，  
全量表包含所有数据，在Hive中名称叫service\_all，增量表包含部分时间段数据，在Hive中名称叫service\_tmp。  
  
  
（1）全量表导入：导出所有数据，只要部分字段，导入到Hive指定表里  
为实现导入Hive功能，需要先配置HCatalog（HCatalog是Hive子模块）的环境变量，/etc/profile中新增：  
export HCAT\_HOME=/home/fulong/Hive/apache-hive-0.13.1-bin/hcatalog  
   
执行以下命令导入数据：  
fulong@FBI006:~/Sqoop/sqoop-1.4.4/bin$ ./sqoop import \\  
> --connect jdbc:oracle:thin:@192.168.0.147:1521:ORCLGBK  --username SP --password fulong \\  
> --table OMP\_SERVICE \\  
> --columns "SERVICE\_CODE,SERVICE\_NAME,SERVICE\_PROCESS,CREATE\_TIME,ENABLE\_ORG,ENABLE\_PLATFORM,IF\_DEL" \\  
> --hive-import --hive-table SERVICE\_ALL  
   
注意：用户名必须大写  
   
（2）增量表导入：只导出所需时间范围内的数据，只要部分字段，导入到Hive指定表里  
使用以下命令导入数据：  
fulong@FBI006:~/Sqoop/sqoop-1.4.4/bin$ ./sqoop import \\  
> --connect jdbc:oracle:thin:@192.168.0.147:1521:ORCLGBK  --username SP --password fulong \\  
> --table OMP\_SERVICE \\  
> --columns "SERVICE\_CODE,SERVICE\_NAME,SERVICE\_PROCESS,CREATE\_TIME,ENABLE\_ORG,ENABLE\_PLATFORM,IF\_DEL" \\  
> --where "CREATE\_TIME > to\_date('2012/12/4 17:00:00','yyyy-mm-dd hh24:mi:ss') and CREATE\_TIME < to\_date('2012/12/4 18:00:00','yyyy-mm-dd hh24:mi:ss')" \\  
> --hive-import --hive-overwrite --hive-table SERVICE\_TMP  
   
注意：  
由于使用了--hive-overwrite参数，所以该语句可反复执行，往service\_tmp表中覆盖插入最新的增量数据；  
Sqoop还支持使用复杂Sql语句查询数据导入，  
相亲参见http://sqoop.apache.org/docs/1.4.4/SqoopUserGuide.html的“7.2.3.Free-form Query Imports”章节  
（3）验证导入结果：列出所有表，统计行数，查看表结构  
hive> show tables;  
OK  
searchlog  
searchlog\_tmp  
service\_all  
service\_tmp  
Time taken: 0.04 seconds, Fetched: 4 row(s)  
hive> select count(\*) from service\_all;  
Total jobs = 1  
Launching Job 1 out of 1  
Number of reduce tasks determined at compile time: 1  
In order to change the average load for a reducer (in bytes):  
  set hive.exec.reducers.bytes.per.reducer=<number>  
In order to limit the maximum number of reducers:  
  set hive.exec.reducers.max=<number>  
In order to set a constant number of reducers:  
  set mapreduce.job.reduces=<number>  
Starting Job = job\_1407233914535\_0013, Tracking URL = http://FBI003:8088/proxy/application\_1407233914535\_0013/  
Kill Command = /home/fulong/Hadoop/hadoop-2.2.0/bin/hadoop job  -kill job\_1407233914535\_0013  
Hadoop job information for Stage-1: number of mappers: 3; number of reducers: 1  
2014-08-21 16:51:47,389 Stage-1 map = 0%,  reduce = 0%  
2014-08-21 16:51:59,816 Stage-1 map = 33%,  reduce = 0%, Cumulative CPU 1.36 sec  
2014-08-21 16:52:01,996 Stage-1 map = 67%,  reduce = 0%, Cumulative CPU 2.45 sec  
2014-08-21 16:52:07,877 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 3.96 sec  
2014-08-21 16:52:17,639 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 5.29 sec  
MapReduce Total cumulative CPU time: 5 seconds 290 msec  
Ended Job = job\_1407233914535\_0013  
MapReduce Jobs Launched:  
Job 0: Map: 3  Reduce: 1   Cumulative CPU: 5.46 sec   HDFS Read: 687141 HDFS Write: 5 SUCCESS  
Total MapReduce CPU Time Spent: 5 seconds 460 msec  
OK  
6803  
Time taken: 59.386 seconds, Fetched: 1 row(s)  
hive> select count(\*) from service\_tmp;  
Total jobs = 1  
Launching Job 1 out of 1  
Number of reduce tasks determined at compile time: 1  
In order to change the average load for a reducer (in bytes):  
  set hive.exec.reducers.bytes.per.reducer=<number>  
In order to limit the maximum number of reducers:  
  set hive.exec.reducers.max=<number>  
In order to set a constant number of reducers:  
  set mapreduce.job.reduces=<number>  
Starting Job = job\_1407233914535\_0014, Tracking URL = http://FBI003:8088/proxy/application\_1407233914535\_0014/  
Kill Command = /home/fulong/Hadoop/hadoop-2.2.0/bin/hadoop job  -kill job\_1407233914535\_0014  
Hadoop job information for Stage-1: number of mappers: 3; number of reducers: 1  
2014-08-21 16:53:03,951 Stage-1 map = 0%,  reduce = 0%  
2014-08-21 16:53:15,189 Stage-1 map = 67%,  reduce = 0%, Cumulative CPU 2.17 sec  
2014-08-21 16:53:16,236 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 3.38 sec  
2014-08-21 16:53:57,935 Stage-1 map = 100%,  reduce = 22%, Cumulative CPU 3.78 sec  
2014-08-21 16:54:01,811 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 5.34 sec  
MapReduce Total cumulative CPU time: 5 seconds 340 msec  
Ended Job = job\_1407233914535\_0014  
MapReduce Jobs Launched:  
Job 0: Map: 3  Reduce: 1   Cumulative CPU: 5.66 sec   HDFS Read: 4720 HDFS Write: 3 SUCCESS  
Total MapReduce CPU Time Spent: 5 seconds 660 msec  
OK  
13  
Time taken: 75.856 seconds, Fetched: 1 row(s)  
hive> describe service\_all;  
OK  
service\_code            string  
service\_name            string  
service\_process         string  
create\_time             string  
enable\_org              string  
enable\_platform         string  
if\_del                  string  
Time taken: 0.169 seconds, Fetched: 7 row(s)  
hive> describe service\_tmp;  
OK  
service\_code            string  
service\_name            string  
service\_process         string  
create\_time             string  
enable\_org              string  
enable\_platform         string  
if\_del                  string  
Time taken: 0.117 seconds, Fetched: 7 row(s)  
步骤2：通过Hive将“全量表+增量表”合并为“更新后的全量表”，覆盖当前的全量表  
合并新表的逻辑如下：  
整个tmp表进入最终表中  
all表的数据中不包含在tmp表service\_code范围内的数据全部进入新表  
执行以下sql语句可以合并得到更新后的全量表：  
hive> select \* from service\_tmp   
union all   
select a.\* from service\_all a left outer join service\_tmp b on a.service\_code = b.service\_code   
where b.service\_code is null;  
   
我们需要直接将查询结果更新回全量表中：  
hive> insert overwrite table service\_all   
select \* from service\_tmp   
union all select a.\* from service\_all a left outer join service\_tmp b on a.service\_code = b.service\_code   
where b.service\_code is null;  
   
   
注意，将查询结果插入表有以下两类语法：  
INSERT OVERWRITE TABLE tablename1 \[PARTITION (partcol1=val1, partcol2=val2 ...) \[IF NOT EXISTS\]\] select\_statement1 FROM from\_statement;  
INSERT INTO TABLE tablename1 \[PARTITION (partcol1=val1, partcol2=val2 ...)\] select\_statement1 FROM from\_statement;  
INSERT OVERWRITE 将会覆盖现有数据，由于当前场景需要更新全量表，所以使用了覆盖模式；  
INSERT INTO 不会覆盖现有数据，是追加数据  
   
到此为止，Hive中的service\_all表已经更新为最新的数据！  
在真实场景中，需要结合shell+cron实现该过程的定时执行。  
通过Sqoop实现Mysql / Oracle 与HDFS / Hbase互导数据 http://www.linuxidc.com/Linux/2013-06/85817.htm  
  
  
\[Hadoop\] Sqoop安装过程详解 http://www.linuxidc.com/Linux/2013-05/84082.htm  
  
  
用Sqoop进行MySQL和HDFS系统间的数据互导 http://www.linuxidc.com/Linux/2013-04/83447.htm  
  
  
Hadoop Oozie学习笔记 Oozie不支持Sqoop问题解决 http://www.linuxidc.com/Linux/2012-08/67027.htm  
  
  
Hadoop生态系统搭建（hadoop hive hbase zookeeper oozie Sqoop） http://www.linuxidc.com/Linux/2012-03/55721.htm  
  
  
Hadoop学习全程记录——使用Sqoop将MySQL中数据导入到Hive中 http://www.linuxidc.com/Linux/2012-01/51993.htm  
  
  
更多Oracle相关信息见Oracle 专题页面 http://www.linuxidc.com/topicnews.aspx?tid=12  
  
  
本文永久更新链接地址：http://www.linuxidc.com/Linux/2014-09/106282.htm