mysql执行拉链表操作
时间：2020-06-01 22:59:23
收藏：0
阅读：105
拉链表需求：
1.数据量比较大
2.变化的比例和频率比较小，例如客户的住址信息，联系方式等，比如有1千万的用户数据，每天全量存储会存储很多不变的信息，对存储也是浪费，因此可以使用拉链表的算法来节省存储空间
3.拉链历史表，既能反映每个客户不同时间的不同状态，也可查看某个时间点的全量快照信息
拉链表设计
设计的拉链历史表：
反映A客户的状态信息
select * from ods_account where cst_id=‘A‘;
反映20190601历史数据：
select * from ods_account where eff_date<=‘20190601‘ and end_date>‘20190601‘;
反映20190602历史全量数据：
select * from ods_account where eff_date<=‘20190602‘ and end_date>‘20190602‘;
建表：
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‘,‘20‘,‘20190601‘);
insert into src_account values(‘B‘,‘30‘,‘20190601‘);
insert into src_account values(‘C‘,‘50‘,‘20190601‘);
insert into src_account values(‘D‘,‘35‘,‘20190601‘);
insert into src_account values(‘A‘,‘20‘,‘20190602‘);
insert into src_account values(‘B‘,‘50‘,‘20190602‘);
insert into src_account values(‘D‘,‘20‘,‘20190602‘);
insert into src_account values(‘E‘,‘50‘,‘20190602‘);
insert into src_account values(‘A‘,‘30‘,‘20190603‘);
insert into src_account values(‘B‘,‘50‘,‘20190603‘);
insert into src_account values(‘C‘,‘50‘,‘20190603‘);
insert into src_account values(‘D‘,‘20‘,‘20190603‘);
insert into src_account values(‘A‘,‘30‘,‘20190604‘);
insert into src_account values(‘B‘,‘40‘,‘20190604‘);
insert into src_account values(‘C‘,‘30‘,‘20190604‘);
insert into src_account values(‘D‘,‘20‘,‘20190604‘);
insert into src_account values(‘E‘,‘20‘,‘20190604‘);
insert into src_account values(‘F‘,‘20‘,‘20190604‘);
insert into src_account values(‘G‘,‘20‘,‘20190604‘);
开始拉链过程：
#清空增量数据
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=‘99991231‘ 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=‘99991231‘ and exists ( select 1 from delta_account s where etl_flag in (‘I‘,‘D‘,‘A‘) and t.cst_id