mysql执行拉链表操作

拉链表需求：

　　1.数据量比较大

　　2.变化的比例和频率比较小，例如客户的住址信息，联系方式等，比如有1千万的用户数据，每天全量存储会存储很多不变的信息，对存储也是浪费，因此可以使用拉链表的算法来节省存储空间

3.拉链历史表，既能反映每个客户不同时间的不同状态，也可查看某个时间点的全量快照信息

拉链表设计

设计的拉链历史表：

反映A客户的状态信息

select * from ods_account where cst_id='A';

反映20190601历史数据：

 select * from ods_account where eff_date<='' and end_date>'';

反映20190602历史全量数据：

 select * from ods_account where eff_date<='' and end_date>'';

建表：

use edw;

drop table if exists src_account;

create table if not exists src_account(

cst_id varchar(64) comment '客户唯一编号',

bal    float       comment '余额',

date_id varchar(16) comment '日期'

)ENGINE=InnoDB DEFAULT CHARSET=utf8;

alter table src_account add primary key(cst_id,date_id);

drop table if exists delta_account;

create table if not exists delta_account(

cst_id varchar(64) comment '客户唯一编号',

bal    float       comment '余额',

etl_flag varchar(16) comment 'ETL标记'

)ENGINE=InnoDB DEFAULT CHARSET=utf8;

alter table delta_account add primary key(cst_id,etl_flag);

drop table if exists odshis_account;

create table if not exists odshis_account(

cst_id varchar(64) comment '客户唯一编号',

bal    float       comment '余额',

eff_date varchar(16) comment '生效日期',

end_date varchar(16) comment '失效日期',

job_seq_id varchar(16) comment '批次号',

new_job_seq_id varchar(16) comment '最新批次号'

)ENGINE=InnoDB DEFAULT CHARSET=utf8;

alter table odshis_account add primary key(cst_id,new_job_seq_id);

drop table if exists ods_account;

create table if not exists ods_account(

cst_id varchar(64) comment '客户唯一编号',

bal    float       comment '余额',

eff_date varchar(16) comment '生效日期',

end_date varchar(16) comment '失效日期',

job_seq_id varchar(16) comment '批次号'

)ENGINE=InnoDB DEFAULT CHARSET=utf8;

alter table ods_account add primary key(cst_id,eff_date,end_date);

加载原始数据：

delete from src_account;

insert into src_account values('A','','');

insert into src_account values('B','','');

insert into src_account values('C','','');

insert into src_account values('D','','');

insert into src_account values('A','','');

insert into src_account values('B','','');

insert into src_account values('D','','');

insert into src_account values('E','','');

insert into src_account values('A','','');

insert into src_account values('B','','');

insert into src_account values('C','','');

insert into src_account values('D','','');

insert into src_account values('A','','');

insert into src_account values('B','','');

insert into src_account values('C','','');

insert into src_account values('D','','');

insert into src_account values('E','','');

insert into src_account values('F','','');

insert into src_account values('G','','');

开始拉链过程：

 #清空增量数据

   truncate delta_account;

   #加载增量数据(新增)

   insert into delta_account

select t1.cst_id,t1.bal,'I' as etl_flag from

(select * from src_account where date_id = '${job_date_id}') t1

left join

(select * from src_account where date_id = '${before_job_date_id}') t2

on t1.cst_id = t2.cst_id where t2.cst_id is null;

#加载增量数据(删除)

insert into delta_account

select t1.cst_id,t1.bal,'D' as etl_flag from

(select * from src_account where date_id = '${before_job_date_id}') t1

left join

(select * from src_account where date_id = '${job_date_id}') t2

on t1.cst_id = t2.cst_id where t2.cst_id is null;

#加载增量数据(变更前)

insert into delta_account

select t1.cst_id,t1.bal,'A' as etl_flag from

(select * from src_account where date_id = '${job_date_id}') t1

left join

(select * from src_account where date_id = '${before_job_date_id}') t2

on t1.cst_id = t2.cst_id where t2.cst_id is not null

and t1.bal <> t2.bal;

#加载增量数据(变更后)

insert into delta_account

select t1.cst_id,t2.bal,'B' as etl_flag from

(select * from src_account where date_id = '${job_date_id}') t1

left join

(select * from src_account where date_id = '${before_job_date_id}') t2

on t1.cst_id = t2.cst_id where t2.cst_id is not null

and t1.bal <> t2.bal;

#1.重跑:删除已跑入数据

delete from ods_account where job_seq_id = '${job_date_id}';

#2.重跑:从历史表恢复数据

insert into ods_account(cst_id,bal,eff_date,end_date,job_seq_id)

select  cst_id,bal,eff_date,end_date,job_seq_id from odshis_account

where new_job_seq_id = '${job_date_id}';

#3.重跑:删除已跑入历史数据

delete from odshis_account where new_job_seq_id = '${job_date_id}';

#4.备份数据到历史表

insert into odshis_account(cst_id,bal,eff_date,end_date,job_seq_id,new_job_seq_id)

select cst_id,bal,eff_date,end_date,job_seq_id,'${job_date_id}'

from ods_account t

where t.end_date='' and exists ( select 1 from delta_account s

where t.cst_id=s.cst_id );

#5.断链

update ods_account t set end_date='${job_date_id}',job_seq_id = '${job_date_id}' where t.end_date='' and exists ( select 1 from delta_account s where etl_flag in ('I','D','A') and t.cst_id=s.cst_id );

#6.加链

insert into ods_account(cst_id,bal,eff_date,end_date,job_seq_id)

select cst_id,bal,'${job_date_id}' as eff_date,'' as end_date,'${job_date_id}' as job_seq_id from delta_account where etl_flag in ('A','I');

#7.保持数据完整性

insert into ods_account (cst_id,bal,eff_date,end_date,job_seq_id)

select t.cst_id,t.bal,'${job_date_id}','${job_date_id}' as end_date,'${job_date_id}' as job_seq_id from delta_account t  where etl_flag = 'D' and not exists (select 1 from ods_account s

where t.cst_id=s.cst_id)