数仓项目拉链表_拉链表用sql如何实现?用调度工具如何实现?

一、拉链表的功能和使用场景

拉链表专门用于解决在数据仓库中数据发生变化如何实现数据存储的问题，如果直接覆盖历史状态，会导致无法查询历史状态，如果将所有数据单独切片存储，会导致存储大量非更新数据的问题。拉链表的设计是将更新的数据进行状态记录，没有发生更新的数据不进行状态存储，用于存储所有数据在不同时间上的所有状态，通过时间进行标记每个状态的生命周期，查询时，根据需求可以获取指定时间范围状态的数据，默认用9999-12-31等最大值来表示最新状态。

【实现过程】
整体实现过程一般分为三步，
第一步先增量采集所有新增数据【增加的数据和发生变化的数据】放入一张增量表。
第二步创建一张临时表，用于将老的拉链表与增量表进行合并。
第三步，最后将临时表的数据覆盖写入拉链表中。

例如：当前MySQL中的数据

当前Hive数据仓库中拉链表的数据：

step1：增量采集变化数据，放入增量表中

step2：构建临时表，将Hive中的拉链表与临时表的数据进行合并

step3：将临时表的数据覆盖写入拉链表中

二、 拉链表的实现

1、数据准备
创建dw层拉链表

--创建数据库
create database db_zipper;
use db_zipper;

--创建拉链表
create table dw_zipper(
  userid string,
  phone string,
  nick string,
  gender int,
  addr string,
  starttime string,
  endtime string
) row format delimited fields terminated by '\t';
1
2
3
4
5
6
7
8
9
10
11
12
13
14

构建模拟数据：vim /export/data/zipper.txt

001	186xxxx1234	laoda	0	sh	2021-01-01	9999-12-31
002	186xxxx1235	laoer	1	bj	2021-01-01	9999-12-31
003	186xxxx1236	laosan	0	sz	2021-01-01	9999-12-31
004	186xxxx1237	laosi	1	gz	2021-01-01	9999-12-31
005	186xxxx1238	laowu	0	sh	2021-01-01	9999-12-31
006	186xxxx1239	laoliu	1	bj	2021-01-01	9999-12-31
007	186xxxx1240	laoqi	0	sz	2021-01-01	9999-12-31
008	186xxxx1241	laoba	1	gz	2021-01-01	9999-12-31
009	186xxxx1242	laojiu	0	sh	2021-01-01	9999-12-31
010	186xxxx1243	laoshi	1	bj	2021-01-01	9999-12-31
1
2
3
4
5
6
7
8
9
10

加载拉链表数据

--加载模拟数据
load data local inpath '/export/data/zipper.txt' into table dw_zipper;
1
2

查询数据

select userid,nick,addr,starttime,endtime from dw_zipper;
1

2.增量采集
创建ods层增量表

create table ods_zipper_update(
  userid string,
  phone string,
  nick string,
  gender int,
  addr string,
  starttime string,
  endtime string
) row format delimited fields terminated by '\t';
1
2
3
4
5
6
7
8
9

创建模拟数据：vim /export/data/update.txt

008	186xxxx1241	laoba	1	sh	2021-01-02	9999-12-31
011	186xxxx1244	laoshi	1	jx	2021-01-02	9999-12-31
012	186xxxx1245	laoshi	0	zj	2021-01-02	9999-12-31
1
2
3

加载更新数据

load data local inpath '/export/data/update.txt' into table ods_zipper_update;
1

查询数据

select userid,nick,addr,starttime,endtime from ods_zipper_update;
1

3.合并数据
创建临时表

create table tmp_zipper(
  userid string,
  phone string,
  nick string,
  gender int,
  addr string,
  starttime string,
  endtime string
) row format delimited fields terminated by '\t';
1
2
3
4
5
6
7
8
9

合并拉链表与增量表

insert overwrite table tmp_zipper
select
  userid,
  phone,
  nick,
  gender,
  addr,
  starttime,
  endtime
from ods_zipper_update
union all
--查询原来拉链表的所有数据，并将这次需要更新的数据的endTime更改为更新值的startTime
select
  a.userid,
  a.phone,
  a.nick,
  a.gender,
  a.addr,
  a.starttime,
  --如果这条数据没有更新或者这条数据不是要更改的数据，就保留原来的值，否则就改为新数据的开始时间-1
  if(b.userid is null or a.endtime < '9999-12-31', a.endtime ,date_sub(b.starttime,1)) as endtime
from dw_zipper a  left join ods_zipper_update b
on a.userid = b.userid ;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

4 生成最新拉链表
覆盖拉链表

insert overwrite table dw_zipper
select * from tmp_zipper;
1
2