拉链表
没错，就像衣服的拉链一样重要，实用性非常强，使用频率非常高。
拉链表核心思想，像个拉链，支持开链，支持闭链，支持退链，我们通常将最新的数据称为开链数据，历史数据称为闭链数据，拉链表支持历史数据查询，且空间占用较小，但是数据加工处置惩罚较为繁琐，属于时间换空间的设计方式，拉链表一个时间维度中同一个用户只生存一条用户状态。拉链表通常会增长三个技术字段“开始日期starttime、竣事日期endtime、状态标识mark”。通过主键（PK）与历史数据进行对比，判断当前数据与历史数据是否发生厘革，如果发生厘革或者新增则进行相应的开链、闭链操作。
1.1 用户基础表拉链示例
以下使用用户根本信息表对拉链表操作进行深入刨析
2021-01-01用户基础表原始数据如下：
userID为主键，可厘革字段为登录名(userName)、电话号码(phoneNum)、账号状态(status)、近来登录日期(lastLoginDate)。
- setp1：首先设计拉链表的主键，根据原始数据表的表结构，选择userID作为拉链表PK键；
- setp2：选择phoneNum、status作为notPK键，notPK作为对比字段，这里去除了lastLoginDate与dataTime这两个经常变化的字段，如果需要跟踪登陆日期lastLoginDate数据，请使用事件表，notPK键选择原则需要同时满足如下条件：
不能选择每天都发生厘革的无对比意义字段（如dataTime字段，对于原始数据表这个字段是每天都厘革的，所以没有对比意义）；
- setp3：确定数仓技术字段，本次加入数据加工日期etlTimestamp、开始时间startTime、结束时间endTime、标识位mark("i":新增，"u":修改，"d":删除)四个技术字段；
2021-01-02用户拉链表如下：
第一次加载由于拉链表历史数据为空，所以所有数据都为新增数据，标识位标为新增状态"i"，开始时间为数据日期2021/1/1，当前所有数据都为最新数据，多以竣事时间设置为一个较大的时间2999/12/31作为开链时间标识日期。
2021-01-02用户基础表原始数据如下：
使用原始数据表主键userID关联拉链表中开链数据(where endTime=2999/12/31)的userID，对比notPK字段是否相同，我们选择的notPK字段为phoneNum和status，发生厘革的数据：
coolniu2021a0001用户电话号码发生厘革；
coolniu2021a0005用户电话号码发生厘革；
coolniu2021a0007用户状态发生厘革；
新增coolniu2021a0008、coolniu2021a0009、coolniu2021a0010三位用户；
2021-01-03用户拉链表如下：
coolniu2021a0001用户拉链分析
coolniu2021a0001电话号码发生厘革，在拉链表中将竣事日期修改为厘革时间的数据日期2021/1/2
代表coolniu2021a0001用户上一个状态竣事。
又新增了一行coolniu2021a0001的数据
针对这条数据新的状态开始时间startTime为厘革日期的数据日期2021/1/2，竣事时间标识为2999/12/31（代表最新开链状态），标识位"u"（代表这条数据是修改状态）。
coolniu2021a0005用户拉链分析
coolniu2021a0005与coolniu2021a0001用户一样，都是电话号码发生了厘革，拉链处置惩罚方式与coolniu2021a0001一致。
coolniu2021a0007用户拉链分析
coolniu2021a0007用户状态由1酿成了销户状态"2"，拉链表处置惩罚：
coolniu2021a0007账号状态为1的数据，竣事时间修改为数据发生厘革的时间2021/1/2
新增一条coolniu2021a0007记录，标识位mark标识为"d"，开始时间为数据厘革时间2021/1/2，竣事时间修改为2021/1/2，表示该条数据已经闭链。
新增用户拉链分析
coolniu2021a0008、coolniu2021a0009、coolniu2021a0010三个用户为2021-01-02日新增用户，开始时间为2021/1/2，竣事时间为2999/12/31（标识该数据为最新开链状态），mark位标为新增标识"i"。
未厘革用户拉链分析
用户coolniu2021a0002、coolniu2021a0003、coolniu2021a0004、coolniu2021a0006用户notPK数据未厘革，所以保持不变。
2021-01-03用户基础表原始数据如下：
1月3日新增了一条coolniu2021a0011数据。
2021-01-04用户拉链表如下：
由于其他天没有发生厘革，所以在拉链表中保持不变，只新增一条coolniu2021a0011的记录。
1.2 拉链表算法示例
1.2.1 建表
source_table
CREATE TABLE source_table(
userid string,
loginname string,
regiondate string,
phonenum string,
birthday string,
status string,
lastlogindate string)
PARTITIONED BY (datatime string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ‘|’
STORED AS TEXTFILE;
1.2.2 load数据
load data local INPATH ‘/tmp/xinniu/20210101’ into table source_table partition (datatime=‘20210101’);
load data local INPATH ‘/tmp/xinniu/20210102’ into table source_table partition (datatime=‘20210102’);
load data local INPATH ‘/tmp/xinniu/20210103’ into table source_table partition (datatime=‘20210103’);
1.2.3 创建拉链表
CREATE TABLE zip_table(
userid string,
loginname string,
regiondate string,
phonenum string,
birthday string,
status string,
lastlogindate string,
datatime string,
etltimestamp string,
starttime string,
endtime string,
mark string)
1.2.4 创建算法文件sqlfile
vim /tmp/xinniu/sqlfile
填写如下内容
– 创建一张拉链表的备份表 备份拉链表历史开链数据
CREATE TABLE IF NOT EXISTS xinniu.zip_table_bk stored AS orc tblproperties (“orc.compress” = “SNAPPY”) AS
SELECT
userID,
loginName,
regionDate,
phoneNum,
birthday,
status,
lastLoginDate,
dataTime,
etlTimestamp ,
startTime ,
endTime ,
mark
FROM
xinniu.zip_table
WHERE
1 = 0 ;
– 将拉链表历史开链数据插入到bk备份表中
– 卡拉链条件：startTime < to_date(from_unixtime(unix_timestamp(‘ h i v e c o n f : b a t c h d a t e ′ , ′ y y y y M M d d ′ ) ) ) A N D e n d T i m e > = t o d a t e ( f r o m u n i x t i m e ( u n i x t i m e s t a m p ( ′ {hiveconf:batch_date}' ,'yyyyMMdd'))) AND endTime >= to_date(from_unixtime(unix_timestamp(' hiveconf:batchdate′,′yyyyMMdd′)))ANDendTime>=todate(fromunixtime(unixtimestamp(′{hiveconf:batch_date}’ ,‘yyyyMMdd’)))
– 备份表非空判断：(select count(1) from xinniu.zip_table_bk limit 1) = 0 判断备份表非空才插入 此处必须判空 不能使用drop或者truncate的方式清空备份表 会导致失败重跑时丢数
INSERT
INTO
xinniu.zip_table_bk
SELECT
userID,
loginName,
regionDate,
phoneNum,
birthday,
status,
lastLoginDate,
dataTime,
etlTimestamp ,
startTime ,
endTime ,
mark
FROM
xinniu.zip_table
join (select count(1) cnt from xinniu.zip_table_bk limit 1) b
WHERE
startTime < to_date(from_unixtime(unix_timestamp(‘ h i v e c o n f : b a t c h d a t e ′ , ′ y y y y M M d d ′ ) ) ) A N D e n d T i m e > = t o d a t e ( f r o m u n i x t i m e ( u n i x t i m e s t a m p ( ′ {hiveconf:batch_date}' ,'yyyyMMdd'))) AND endTime >= to_date(from_unixtime(unix_timestamp(' hiveconf:batchdate′,′yyyyMMdd′)))ANDendTime>=todate(fromunixtime(unixtimestamp(′{hiveconf:batch_date}’ ,‘yyyyMMdd’)))
AND b.cnt = 0
;
– 创建拉链表闭链数据备份表bf
CREATE TABLE IF NOT EXISTS xinniu.zip_table_bf stored AS orc tblproperties (“orc.compress” = “SNAPPY”) AS
SELECT
userID,
loginName,
regionDate,
phoneNum,
birthday,
status,
lastLoginDate,
dataTime,
etlTimestamp ,
startTime ,
endTime ,
mark
FROM
xinniu.zip_table
WHERE
1 = 0 ;
– 备份拉链表中历史闭链数据 卡拉链条件：endTime < to_date(to_timestamp(‘${hiveconf:batch_date}’ , ‘yyyyMMdd’))
– 备份表非空判断：(select count(1) from xinniu.zip_table_bf limit 1) = 0 同上，不能使用drop或者truncate的方式清空备份表
INSERT
INTO
xinniu.zip_table_bf
SELECT
userID,
loginName,
regionDate,
phoneNum,
birthday,
status,
lastLoginDate,
dataTime,
etlTimestamp ,
startTime ,
endTime ,
mark
FROM
xinniu.zip_table
join
(
SELECT
count(1) cnt
FROM
xinniu.zip_table_bf
LIMIT 1) b
WHERE
endTime < to_date(from_unixtime(unix_timestamp(‘${hiveconf:batch_date}’ ,‘yyyyMMdd’)))
AND b.cnt = 0 ;
– 中间加工表清空
DROP TABLE IF EXISTS xinniu.zip_table_nw;
DROP TABLE IF EXISTS xinniu.zip_table_od;
– 创建中间表 新增厘革修改中间表
CREATE TABLE IF NOT EXISTS xinniu.zip_table_nw stored AS orc tblproperties (“orc.compress” = “SNAPPY”) AS
SELECT
userID,
loginName,
regionDate,
phoneNum,
birthday,
status,
lastLoginDate,
dataTime,
etlTimestamp ,
startTime ,
endTime ,
mark
FROM
xinniu.zip_table
WHERE
0 = 1;
– 创建中间表 未厘革中间表
CREATE TABLE IF NOT EXISTS xinniu.zip_table_od stored AS orc tblproperties (“orc.compress” = “SNAPPY”) AS
SELECT
userID,
loginName,
regionDate,
phoneNum,
birthday,
status,
lastLoginDate,
dataTime,
etlTimestamp ,
startTime ,
endTime ,
mark
FROM
xinniu.zip_table
WHERE
0 = 1;
– 新增、修改、删除厘革数据插入厘革中间表nw
– 原始数据表与拉链表进行full join关联，通过主键is_pk关联 根据差别情况生成对应的startTime与endTime及mark三个技术字段
– 字段值选择原始表与目标表的非空字段值 nvl(n.@{source_column_names}, o.@{xinniu.zip_table_column_names})
– 本逻辑中mark分为I、D两种，新增与修改为I，删除为D
INSERT
INTO
TABLE xinniu.zip_table_nw
SELECT
nvl(n.userID,o.userID) ,
nvl(n.loginName,o.loginName) ,
nvl(n.regionDate,o.regionDate) ,
nvl(n.phoneNum,o.phoneNum) ,
nvl(n.birthday,o.birthday) ,
nvl(n.status,o.status) ,
nvl(n.lastLoginDate,o.lastLoginDate) ,
nvl(n.dataTime,o.dataTime) ,
current_date AS etlTimestamp ,
CASE
WHEN n.dataTime IS NULL THEN o.startTime
ELSE to_date(from_unixtime(unix_timestamp(‘ h i v e c o n f : b a t c h d a t e ′ , ′ y y y y M M d d ′ ) ) ) E N D A S s t a r t T i m e , C A S E W H E N n . d a t a T i m e I S N U L L T H E N t o d a t e ( f r o m u n i x t i m e ( u n i x t i m e s t a m p ( ′ {hiveconf:batch_date}' ,'yyyyMMdd'))) END AS startTime , CASE WHEN n.dataTime IS NULL THEN to_date(from_unixtime(unix_timestamp(' hiveconf:batchdate′,′yyyyMMdd′)))ENDASstartTime,CASEWHENn.dataTimeISNULLTHENtodate(fromunixtime(unixtimestamp(′{hiveconf:batch_date}’ ,‘yyyyMMdd’)))
ELSE to_date(from_unixtime(unix_timestamp(‘29991231’ ,‘yyyyMMdd’)))
END AS endTime ,
CASE
WHEN ( n.userID is null ) THEN ‘D’
ELSE ‘I’
END AS mark
FROM
(
SELECT
userID,
loginName,
regionDate,
phoneNum,
birthday,
status,
lastLoginDate,
dataTime
FROM
xinniu.source_table
WHERE
dataTime = ‘${hiveconf:batch_date}’ ) n
FULL JOIN xinniu.zip_table_bk o ON
o.userID = n.userID
WHERE
(
o.userID IS NULL )
OR (
n.userID IS NULL )
OR (
nvl( CAST(o.phoneNum AS string) , ‘’ ) <> nvl( CAST(n.phoneNum AS string) , ‘’ )
OR nvl( CAST(o.status AS string) , ‘’ ) <> nvl( CAST(n.status AS string) , ‘’ )
)
;
– 闭链发生厘革的数据 endTime改为hiveconf:batch_date
– 未厘革数据保持原来状态 新增与修改状态同一"I"
– 发生厘革的endTime逻辑：when n.startTime is not null then to_date(to_timestamp(‘ h i v e c o n f : b a t c h d a t e ′ , ′ y y y y M M d d ′ ) ) − − 没发生厘革的 e n d T i m e 逻辑： w h e n o . e n d T i m e > = t o d a t e ( t o t i m e s t a m p ( ′ {hiveconf:batch_date}' , 'yyyyMMdd')) -- 没发生厘革的endTime逻辑：when o.endTime >= to_date(to_timestamp(' hiveconf:batchdate′,′yyyyMMdd′))−−没发生厘革的endTime逻辑：wheno.endTime>=todate(totimestamp(′{hiveconf:batch_date}’ , ‘yyyyMMdd’)) then to_date(to_timestamp(‘29991231’,‘yyyyMMdd’))
INSERT
INTO
TABLE xinniu.zip_table_od
SELECT
o.userID,
o.loginName,
o.regionDate,
o.phoneNum,
o.birthday,
o.status,
o.lastLoginDate,
o.dataTime,
o.etlTimestamp ,
o.startTime ,
CASE
WHEN n.startTime IS NOT NULL THEN to_date(from_unixtime(unix_timestamp(‘ h i v e c o n f : b a t c h d a t e ′ , ′ y y y y M M d d ′ ) ) ) W H E N o . e n d T i m e > = t o d a t e ( f r o m u n i x t i m e ( u n i x t i m e s t a m p ( ′ {hiveconf:batch_date}' ,'yyyyMMdd'))) WHEN o.endTime >= to_date(from_unixtime(unix_timestamp(' hiveconf:batchdate′,′yyyyMMdd′)))WHENo.endTime>=todate(fromunixtime(unixtimestamp(′{hiveconf:batch_date}’ ,‘yyyyMMdd’)))
THEN to_date(from_unixtime(unix_timestamp(‘29991231’ ,‘yyyyMMdd’)))
ELSE o.endTime
END AS endTime ,
‘I’ AS mark
FROM
xinniu.zip_table_bk o
LEFT JOIN xinniu.zip_table_nw n ON
o.userID = n.userID
WHERE
nvl(n.endTime,to_date(from_unixtime(unix_timestamp(‘29991231’ ,‘yyyyMMdd’)))) <> to_date(from_unixtime(unix_timestamp(‘${hiveconf:batch_date}’ ,‘yyyyMMdd’)))
;
– 清空拉链表
TRUNCATE TABLE xinniu.zip_table;
– 插入数据到拉链表
INSERT
INTO
TABLE xinniu.zip_table
SELECT
*
FROM
xinniu.zip_table_nw
UNION ALL
SELECT
*
FROM
xinniu.zip_table_od
UNION ALL
SELECT
*
FROM
xinniu.zip_table_bf ;
– 清空暂时表
DROP TABLE xinniu.zip_table_bk;
DROP TABLE xinniu.zip_table_bf;
DROP TABLE xinniu.zip_table_nw;
DROP TABLE xinniu.zip_table_od;
1.2.5 执行跑批任务
hive -hiveconf batch_date=20210101 -f /tmp/xinniu/sqlfile && hive -hiveconf batch_date=20210102 -f /tmp/xinniu/sqlfile && hive -hiveconf batch_date=20210103 -f /tmp/xinniu/sqlfile
1.3 效果验证
1.3.1 全表验证
select * from xinniu.zip_table;
1.3.2 卡拉链，查询历史某一节点数据
select * from xinniu.zip_table where startTime<=‘2021-01-01’ and endTime>‘2021-01-01’;
select * from xinniu.zip_table where startTime<=‘2021-01-02’ and endTime>‘2021-01-02’;
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
