哪里有做网站推广,当前最好用的wordpress主题,网站开发技术留言,安徽定制型网站建设推广背景 拉链表是一种数据模型#xff0c;主要是针对数据仓库设计中表存储数据的方式而定义的#xff1b;顾名思义#xff0c;所谓拉链表#xff0c;就是记录历史。记录一个事务从开始一直到当前状态的所有变化的信息。 拉链表可以避免按每一天存储所有记录造成的海量存储问题…背景 拉链表是一种数据模型主要是针对数据仓库设计中表存储数据的方式而定义的顾名思义所谓拉链表就是记录历史。记录一个事务从开始一直到当前状态的所有变化的信息。 拉链表可以避免按每一天存储所有记录造成的海量存储问题同时也是处理缓慢变化数据SCD2的一种常见方式。 应用场景 现假设有如下场景一个企业拥有5000万会员信息每天有20万会员资料变更需要在数仓中记录会员表的历史变化以备分析使用即每天都要保留一个快照供查询反映历史数据的情况。 在此场景中需要反映5000万会员的历史变化如果保留快照存储两年就需要2X365X5000W条数据存储空间数据量为365亿如果存储更长时间则无法估计需要的存储空间。而利用拉链算法存储每日只向历史表中添加新增和变化的数据每日不过20万条存储4年也只需要3亿存储空间。
实现步骤 在拉链表中每一条数据都有一个生效日期(effective_date)和失效日期(expire_date)。假设在一个用户表中在2019年11月8日新增了两个用户如下表所示则这两条记录的生效时间为当天由于到2019年11月8日为止,这两条就还没有被修改过所以失效时间为一个给定的比较大的值比如3000-12-31
member_idphonenocreate_timeupdate_time10001133000000012019-11-083000-12-3110002135000000022019-11-083000-12-31 第二天(2019-11-09)用户10001被删除了用户10002的电话号码被修改成13600000002.为了保留历史状态用户10001的失效时间被修改为2019-11-09用户10002则变成了两条记录如下表所示
member_idphonenocreate_timeupdate_time10001133000000012019-11-082019-11-0910002135000000022019-11-082019-11-0910002136000000022019-11-093000-12-31 第三天(2019-11-10),又新增了用户10003则用户表数据如小表所示
member_idphonenocreate_timeupdate_time10001133000000012019-11-082019-11-0910002135000000022019-11-082019-11-0910002136000000022019-11-093000-12-3110003133000000062019-11-103000-12-31 如果要查询最新的数据那么只要查询失效时间为3000-12-31的数据即可如果要查11月8号的历史数据则筛选生效时间 2019-11-08并且失效时间2019-11-08的数据即可。如果查询11月9号的数据那么筛选条件则是生效时间2019-11-09并且失效时间2019-11-09
表结构 MySQL源member表
CREATE TABLE member(member_id VARCHAR ( 64 ),phoneno VARCHAR ( 20 ),create_time datetime,update_time datetime ); ODS层增量表member_delta,每天一个分区
CREATE TABLE member_delta(member_id string,phoneno string,create_time string,update_time string)
PARTITIONED BY (DAY string);临时表
CREATE TABLE member_his_tmp(member_id string,phoneno string,effective_date date,expire_date date); DW层历史拉链表
CREATE TABLE member_his(member_id string,phoneno string,effective_date date,expire_date date); Demo数据准备
2019-11-08的数据为
member_idphonenocreate_timeupdate_time10001135000000012019-11-08 14:47:552019-11-08 14:47:5510002135000000022019-11-08 14:48:332019-11-08 14:48:3310003135000000032019-11-08 14:48:532019-11-08 14:48:5310004135000000042019-11-08 14:49:022019-11-08 14:49:02 2019-11-09的数据为其中蓝色代表新增数据红色代表修改的数据
member_idphonenocreate_timeupdate_time10001135000000012019-11-08 14:47:552019-11-08 14:47:5510002136000000022019-11-08 14:48:332019-11-09 14:48:3310003135000000032019-11-08 14:48:532019-11-08 14:48:5310004135000000042019-11-08 14:49:022019-11-08 14:49:0210005135000000052019-11-09 08:54:032019-11-09 08:54:0310006135000000062019-11-09 09:54:252019-11-09 09:54:25 2019-11-10的数据其中蓝色代表新增数据红色代表修改的数据 member_idphonenocreate_timeupdate_time10001135000000012019-11-08 14:47:552019-11-08 14:47:5510002136000000022019-11-08 14:48:332019-11-09 14:48:3310003135000000032019-11-08 14:48:532019-11-08 14:48:5310004136000000042019-11-08 14:49:022019-11-10 14:49:0210005135000000052019-11-09 08:54:032019-11-09 08:54:0310006135000000062019-11-09 09:54:252019-11-09 09:54:2510007135000000072019-11-10 17:41:492019-11-10 17:41:49 全量初始装载 在启用拉链表时先对其进行初始装载比如以2019-11-08为开始时间那么将MySQL源表全量抽取到ODS层member_delta表的2018-11-08的分区中然后初始装载DW层的拉链表member_his
INSERT overwrite TABLE member_his
SELECTmember_id,phoneno,to_date ( create_time ) AS effective_date,3000-12-31
FROM
member_delta
WHERE
DAY 2019-11-08查询初始的历史拉链表数据 增量抽取数据 每天从源系统member表中将前一天的增量数据抽取到ODS层的增量数据表member_delta对应的分区中。这里的增量需要通过member表中的创建时间和修改时间来确定或者使用sqoop job监控update时间来进行增联抽取。比如本案例中2019-11-09和2019-11-10为两个分区分别存储了2019-11-09和2019-11-10日的增量数据。2019-11-09分区的数据为: 2019-11-10分区的数据为 增量刷新历史拉链数据 2019-11-09增量刷新历史拉链表将数据放进临时表
INSERT overwrite TABLE member_his_tmp
SELECT *
FROM(
-- 2019-11-09增量数据代表最新的状态该数据的生效时间是2019-11-09过期时间为3000-12-31
-- 这些增量的数据需要被全部加载到历史拉链表中
SELECT member_id,phoneno,2019-11-09 effective_date,3000-12-31 expire_dateFROM member_deltaWHERE DAY2019-11-09UNION ALL
-- 用当前为生效状态的拉链数据去left join 增量数据
-- 如果匹配得上则表示该数据已发生了更新
-- 此时需要将发生更新的数据的过期时间更改为当前时间.
-- 如果匹配不上则表明该数据没有发生更新此时过期时间不变
SELECT a.member_id,a.phoneno,a.effective_date,if(b.member_id IS NULL, to_date(a.expire_date), to_date(b.day)) expire_dateFROM(SELECT *FROM member_hisWHERE expire_date3000-12-31) aLEFT JOIN(SELECT *FROM member_deltaWHERE DAY2019-11-09) b ON a.member_idb.member_id)his将数据覆盖到历史拉链表
INSERT overwrite TABLE member_his
SELECT *
FROM member_his_tmp查看历史拉链表 2019-11-10增量刷新历史拉链表 将数据放进临时表
INSERT overwrite TABLE member_his_tmp
SELECT *
FROM
(
-- 2019-11-10增量数据代表最新的状态该数据的生效时间是2019-11-10过期时间为3000-12-31
-- 这些增量的数据需要被全部加载到历史拉链表中
SELECT member_id,phoneno,2019-11-10 effective_date,3000-12-31 expire_dateFROM member_deltaWHERE DAY2019-11-10UNION ALL
-- 用当前为生效状态的拉链数据去left join 增量数据
-- 如果匹配得上则表示该数据已发生了更新
-- 此时需要将发生更新的数据的过期时间更改为当前时间.
-- 如果匹配不上则表明该数据没有发生更新此时过期时间不变
SELECT a.member_id,a.phoneno,a.effective_date,if(b.member_id IS NULL, to_date(a.expire_date), to_date(b.day)) expire_dateFROM(SELECT *FROM member_hisWHERE expire_date3000-12-31) aLEFT JOIN(SELECT *FROM member_deltaWHERE DAY2019-11-10) b ON a.member_idb.member_id)his 查看历史拉链表 将以上脚本封装成shell调度的脚本
#!/bin/bash #如果是输入的日期按照取输入日期如果没输入日期取当前时间的前一天 if [ -n $1 ] ;then do_date$1 else do_datedate -d -1 day %F fi sql INSERT overwrite TABLE member_his_tmp SELECT * FROM ( -- 2019-11-10增量数据代表最新的状态该数据的生效时间是2019-11-10过期时间为3000-12-31 -- 这些增量的数据需要被全部加载到历史拉链表中 SELECT member_id, phoneno, $do_date effective_date, 3000-12-31 expire_date FROM member_delta WHERE DAY$do_date UNION ALL -- 用当前为生效状态的拉链数据去left join 增量数据 -- 如果匹配得上则表示该数据已发生了更新 -- 此时需要将发生更新的数据的过期时间更改为当前时间. -- 如果匹配不上则表明该数据没有发生更新此时过期时间不变 SELECT a.member_id, a.phoneno, a.effective_date, if(b.member_id IS NULL, to_date(a.expire_date), to_date(b.day)) expire_date FROM (SELECT * FROM member_his WHERE expire_date3000-12-31) a LEFT JOIN (SELECT * FROM member_delta WHERE DAY$do_date) b ON a.member_idb.member_id)his; $hive -e $sql 如需获取更多资料您可以下载知识星球app并搜索加入‘数据要素X’。
