大数据Hive拉链表的设计与实现_hive拉链表的实现过程

---

想学习架构师构建流程请跳转：Java架构师系统架构设计

1 数据同步问题

1.1 数据同步的场景

Hive在实际工作中主要用于构建离线数据仓库，定期的从各种数据源中同步采集数据到Hive中，经过分层转换提供数据应用。例如，每天需要从MySQL中同步最新的订单信息、用户信息、店铺信息等到数据仓库中，进行订单分析、用户分析。
例如：MySQL中有一张用户表：tb_user，每个用户注册完成以后，就会在用户表中新增该用户的信息，记录该用户的id、手机号码、用户名、性别、地址等信息。

每天都会有用户注册，产生新的用户信息，我们每天都需要将MySQL中的用户数据同步到Hive数据仓库中，在做用户分析时，需要对用户的信息做统计分析，例如统计新增用户的个数、总用户个数、用户性别分布、地区分布、运营商分布等指标。

1.2 数据同步的问题

在实现数据仓库数据同步的过程中，我们必须保证Hive中的数据与MySQL中的数据是一致的，这样才能确保我们最终分析出来的结果是准确的，没有问题的，但是在实现同步的过程中，这里会面临一个问题：如果MySQL中的数据发生了修改，Hive中如何存储被修改的数据？
例如以下情况

➢ 2021-01-01：MySQL中有10条用户信息

➢ 2021-01-02：Hive进行数据分析，将MySQL中的数据同步
⚫ 2021-01-02：MySQL中新增2条用户注册数据，并且有1条用户数据发生更新
◼ 新增两条用户数据011和012
◼ 008的addr发生了更新，从gz更新为sh
⚫ 2021-01-03：Hive需要对2号的数据进行同步更新处理
⚫ 问题：新增的数据会直接加载7.1.3 解决方案
➢ 方案一：在Hive中用新的addr覆盖008的老的addr，直接更新
到Hive表中，但是更新的数据如何存储在Hive表中？
➢ 优点：实现最简单，使用起来最方便
➢ 缺点：没有历史状态，008的地址是1月2号在sh，但是1月2号之前是在gz的，如果要查询008的1月2号之前的addr就无法查询，也不能使用sh代替

➢ 方案二：每次数据改变，根据日期构建一份全量的快照表，每天一张表
➢ 2021-01-02：Hive中有一张表tb_user_2021-01-02

➢ 2021-01-03：Hive中有一张表tb_user_2021-01-03
➢ 优点：记录了所有数据在不同时间的状态
➢ 缺点：冗余存储了很多没有发生变化的数据，导致存储的数据量过大

➢ 方案三：构建拉链表，通过时间标记发生变化的数据的每种状态的时间周期

2 拉链表的设计

2.1 功能与应用场景

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

2.2 实现过程

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

当前Hive数据仓库中拉链表的数据：
⚫ step1：增量采集变化数据，放入增量表中
⚫ step2：构建临时表，将Hive中的拉链表与临时表的数据进行合并
⚫ step3：将临时表的数据覆盖写入拉链表中

3 拉链表的实现

3.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';

➢	构建模拟数据：vim /export/data/zipper.txt
```sql
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
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

➢ 加载拉链表数据
–加载模拟数据
load data local inpath ‘/export/data/zipper.txt’ into table dw_zipper;

➢ 查询数据
select userid,nick,addr,starttime,endtime from dw_zipper;

3.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

➢ 加载更新数据
load data local inpath ‘/export/data/update.txt’ into table ods_zipper_update;

➢ 查询数据
select userid,nick,addr,starttime,endtime from ods_zipper_update;

3.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
10

➢ 合并拉链表与增量表

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

3.4 生成最新拉链表

➢ 覆盖拉链表

insert overwrite table dw_zipper
select * from tmp_zipper;
1
2