etl全量增量抽取方式

我们在大数据进行数据抽取&＃xff08;入湖&＃xff09;中离不开从源系统的数据库同步数据到大数据平台&＃xff08;hive&＃xff09;中。我的平台&＃xff08;带有调度工具&＃xff09;&＃xff1a;联想大数据平&＃xff08;LeapHD&＃xff09;。

同步的方式有两种&＃xff1a;

1.增量抽取&＃xff1a;即每次抽取表的全部数据然后覆盖表的全部数据。频率一般每天抽取或者每周每月抽取&＃xff0c;具体看业务需求。

1.1在hive中建一张结构与源表的相同表
eg:
源表&＃xff08;mysql&＃xff09;的结构&＃xff1a;
表名&＃xff1a;user 字段为&＃xff1a;user_id int,user_name varchar,create_time datetime,phone varchar ;
hive建表&＃xff1a;

create table ods.user_id (
etl_date string COMMENT &＃39;入湖时间&＃39;,
user_id string COMMENT &＃39;客户id&＃39;,
user_name string COMMENT &＃39;客户姓名&＃39;,
create_time string COMMENT &＃39;创建时间&＃39;,
phone string COMMENT &＃39;电话&＃39;
)comment &＃39;客户信息表&＃39;
row format delimited fields terminated by &＃39;\001&＃39; --换行方式

1.2 同步 用sqoop 或者用kettle工具同步。我用的是联想平台的自带的etl工具。

抽取全量的源表&＃xff0c;进行任务调度的设置&＃xff0c;是每天或者每周每月&＃xff08;看业务需求&＃xff09;。

select
&＃39;${now}&＃39; etl_time,
user_id ,
user_name ,
create_time ,
phone
from dd.user

导入到创建的hive表

ods.user_id

2.增量导入

增量方式的话我处理的方法是创建分区表&＃xff0c;按照同步方式建立相应的分区。比如在单表数据量较大&＃xff08;比如&＃xff1a;100万以上&＃xff09;的我会按照日分区&＃xff0c;数据量较小的按照月分区。

2.1日分区

2.1.1 创建日分区表

create table ods.user_id (
etl_date string COMMENT &＃39;入湖时间&＃39;,
user_id string COMMENT &＃39;客户id&＃39;,
user_name string COMMENT &＃39;客户姓名&＃39;,
create_time string COMMENT &＃39;创建时间&＃39;,
phone string COMMENT &＃39;电话&＃39;
)comment &＃39;客户信息表&＃39;
partitioned by (day_id string comment &＃39;日分区&＃39;)
row format delimited fields terminated by &＃39;\001&＃39; --换行方式

2.1.2 进行表的初始化数据&＃xff08;跟全量的操作差不多&＃xff0c;多了分区&＃xff09;

全量抽取

select
&＃39;${now}&＃39; etl_time,
user_id ,
user_name ,
create_time ,
phone
from dd.user

导入到创建的hive表指定临时分区

ods.user_id[day_id&＃61;&＃39;1&＃39;] 

2.1.3 将临时分区导入到动态分区

insert overwrite table ods.user_id partition(day_id)
select
etl_time ,
user_id ,
user_name ,
create_time ,
phone ,
substr(create_time,1,10) as day_id --日期格式为 YYYY-MM-DD
from ods.user_id partition
where day_id&＃61;&＃39;1&＃39;

2.1.4 删除表的临时分区

alter table ods.user_id drop partition (day_id&＃61;&＃39;1&＃39;);

2.1.5 完成初始化后就是增量导入增量数据

设置把昨天的日期设置成变量&＃xff1a;{last_day} &＃61; {now} - 1

 抽取昨天的数据

select
&＃39;${now}&＃39; etl_time ,
user_id ,
user_name ,
create_time ,
phone
from dd.user
where substr(create_time,1,10)&＃61;&＃39;{last_day}&＃39;

同理&＃xff0c;抽数放到临时分区

ods.user_id[day_id&＃61;&＃39;000&＃39;] 

2.1.6 将临时分区换成动态分区

insert overwrite table ods.user_id partition(day_id)
select
etl_time ,
user_id ,
user_name ,
create_time ,
phone ,
substr(create_time,1,10) as day_id --日期格式为 YYYY-MM-DD
from ods.user_id partition
where day_id&＃61;&＃39;000&＃39;

2.1.7 删除临时分区

alter table ods.user_id drop partition (day_id&＃61;&＃39;000&＃39;);

这样就完成每天的抽数定时任务。