大数据-246 离线数仓 - 电商分析 拉链表的分析与构建与回滚 ...

点一下关注吧！！！非常感谢！！持续更新！！！

Java篇开始了！

目前开始更新 MyBatis，一起深入浅出！
目前已经更新到了：

• Hadoop（已更完）
  
• HDFS（已更完）
  
• MapReduce（已更完）
  
• Hive（已更完）
  
• Flume（已更完）
  
• Sqoop（已更完）
  
• Zookeeper（已更完）
  
• HBase（已更完）
  
• Redis （已更完）
  
• Kafka（已更完）
  
• Spark（已更完）
  
• Flink（已更完）
  
• ClickHouse（已更完）
  
• Kudu（已更完）
  
• Druid（已更完）
  
• Kylin（已更完）
  
• Elasticsearch（已更完）
  
• DataX（已更完）
  
• Tez（已更完）
  
• 数据挖掘（已更完）
  
• Prometheus（已更完）
  
• Grafana（已更完）
  
• 离线数仓（正在更新…）
  

章节内容

上节我们完成了如下的内容：

• 电商分析 缓慢变化维
  
• 拉链表 SCD Slowly Changing Dimensions
  

拉链表的实现

userinfo（分区表） => userid、mobile、regdate => 每日变动的数据（修改的+新增的）/ 历史数据（第一天）
userhis（拉链表）=> 多个两个字段 start_date / end_date
拉链表（Zipper Table）

拉链表是一种数据库计划模式，用于跟踪数据随时间的变化，同时保持高效的查询性能。这种模式广泛应用于数据仓库和数据分析场景，因为它能够很好地记录历史数据的变化环境。
拉链表的基本概念

拉链表的核心思想是将每条记录的有效时间范围存储起来，通过“拉链”方式记录版本变化。每一条记录都包罗以下关键信息：

• 开始时间（Start Date/Effective Date）：表示这条记录的生效时间。
  
• 竣事时间（End Date/Expiration Date）：表示这条记录的失效时间。
  
• 是否当前有效（Is Current）：表示这条记录是否为最新版本（通常通过标志位存储，如1表示当前记录，0表示历史记录）。
  

工作原理

• 新增数据：当有新数据插入时，系统创建一条新记录，设置其开始时间为当前时间，竣事时间为一个默认的最大时间（如9999-12-31），同时将is_current字段设为1。
  
• 更新数据：起首将现有的有效记录的竣事时间更新为当前时间，表示它的有效期竣事，同时将is_current标志设为0。
  
• 然后插入一条新的记录，表示更新后的版本，开始时间为当前时间，竣事时间为默认最大时间，is_current标志为1。
  
• 删除数据：一般通过逻辑删除方式（更新竣事时间和is_current字段）实现，而不是直接物理删除。
  

预备数据

这里的数据刚才已经全部都写入进去了

1. -- 1、userinfo初始化（2020-06-20）。获取历史数据
2. 001,13551111111,2020-03-01,2020-06-20
3. 002,13561111111,2020-04-01,2020-06-20
4. 003,13571111111,2020-05-01,2020-06-20
5. 004,13581111111,2020-06-01,2020-06-20

复制代码

初始化拉链表

将2020-06-20的数据写入到表中

1. -- 2、初始化拉链表（2020-06-20）。userinfo => userhis
2. INSERT OVERWRITE TABLE test.userhis
3. SELECT
4.     userid,
5.     mobile,
6.     regdate,
7.     dt AS start_date,
8.     '9999-12-31' AS end_date
9. FROM
10.     test.userinfo
11. WHERE
12.     dt = '2020-06-20';

复制代码

实行结果如下所示：

继续预备数据

这批数据也已经写入了：

1. -- 3、次日新增数据（2020-06-21）；获取新增数据
2. 002,13562222222,2020-04-01,2020-06-21
3. 004,13582222222,2020-06-01,2020-06-21
4. 005,13552222222,2020-06-21,2020-06-21

复制代码

构建拉链表

1. -- 4、构建拉链表(userhis)（2020-06-21）【核心】 userinfo(2020-06-21) + userhis => userhis
2. -- userinfo: 新增数据
3. -- userhis：历史数据
4. -- 第一步：处理新增数据【userinfo】（处理逻辑与加载历史数据类似）
5. SELECT
6.     userid,
7.     mobile,
8.     regdate,
9.     dt AS start_date,
10.     '9999-12-31' AS end_date
11. FROM
12.     test.userinfo
13. WHERE
14.     dt = '2020-06-21';
16. -- 第二步：处理历史数据【userhis】（历史包括两部分：变化的、未变化的）
17. -- 变化的：start_date:不变；end_date：传入日期-1
18. -- 未变化的：不做处理
20. -- 观察数据
21. SELECT
22.     A.userid,
23.     B.userid,
24.     B.mobile,
25.     B.regdate,
26.     B.start_date,
27.     B.end_date
28. FROM
29.     (SELECT *
30.      FROM test.userinfo
31.      WHERE dt = '2020-06-21') A
32. RIGHT JOIN
33.     test.userhis B
34. ON
35.     A.userid = B.userid;
37. -- 编写SQL，处理历史数据
38. SELECT
39.     B.userid,
40.     B.mobile,
41.     B.regdate,
42.     B.start_Date,
43.     CASE
44.         WHEN B.end_date = '9999-12-31' AND A.userid IS NOT NULL
45.         THEN DATE_ADD('2020-06-21', INTERVAL -1 DAY)
46.         ELSE B.end_date
47.     END AS end_date
48. FROM
49.     (SELECT * FROM test.userinfo WHERE dt = '2020-06-21') A
50. RIGHT JOIN
51.     test.userhis B
52. ON
53.     A.userid = B.userid;
55. -- 最终的处理（新增+历史数据）
56. INSERT OVERWRITE TABLE test.userhis
57. SELECT
58.     userid,
59.     mobile,
60.     regdate,
61.     dt AS start_date,
62.     '9999-12-31' AS end_date
63. FROM
64.     test.userinfo
65. WHERE
66.     dt = '2020-06-21'
67. UNION ALL
68. SELECT
69.     B.userid,
70.     B.mobile,
71.     B.regdate,
72.     B.start_date,
73.     CASE
74.         WHEN B.end_date = '9999-12-31' AND A.userid IS NOT NULL
75.         THEN date_add('2020-06-21', -1)
76.         ELSE B.end_date
77.     END AS end_date
78. FROM
79.     (SELECT * FROM test.userinfo WHERE dt = '2020-06-21') A
80. RIGHT JOIN
81.     test.userhis B
82. ON
83.     A.userid = B.userid;

复制代码

实行过程如下图所示：

拉链表测试脚本

1. vim test_zipper.sh

复制代码

写入的内容如下所示：

1. #!/bin/bash
2. # 加载环境变量
3. source /etc/profile
5. # 判断是否传入日期参数，如果没有则使用前一天的日期
6. if [ -n "$1" ]; then
7.     do_date=$1
8. else
9.     do_date=$(date -d "-1 day" +%F)
10. fi
12. # SQL 语句
13. sql="
14. INSERT OVERWRITE TABLE test.userhis
15. SELECT
16.     userid,
17.     mobile,
18.     regdate,
19.     dt AS start_date,
20.     '9999-12-31' AS end_date
21. FROM
22.     test.userinfo
23. WHERE
24.     dt = '$do_date'
25. UNION ALL
26. SELECT
27.     B.userid,
28.     B.mobile,
29.     B.regdate,
30.     B.start_date,
31.     CASE
32.         WHEN B.end_date = '9999-12-31' AND A.userid IS NOT NULL THEN date_add('$do_date', -1)
33.         ELSE B.end_date
34.     END AS end_date
35. FROM
36.     (SELECT * FROM test.userinfo WHERE dt = '$do_date') A
37. RIGHT JOIN
38.     test.userhis B
39. ON
40.     A.userid = B.userid;
41. "
43. # 执行 Hive SQL
44. hive -e "$sql"

复制代码

拉链表的回滚

由于种种原因必要将拉链表规复到rollback_date那一天的数据，此时有：

• end_date < rollback_date，即竣事日期<回滚日期，表示该行数据在roll_back_date之前产生，这些数据必要原样保留
  
• start_date <= rollback_date <= end_date，即开始日期 <= 回滚日期 <= 竣事日期，这些数据是回滚日期之后产生的，但是必要修改，将end_date改为9999-12-31
  
• 其他数据不用管
  

按照上述方案进行编码：
处理end_date < rollback_date 的数据

1. SELECT
2.     userid,
3.     mobile,
4.     regdate,
5.     start_date,
6.     end_date,
7.     '1' AS tag
8. FROM
9.     test.userhis
10. WHERE
11.     end_date < '2020-06-22';

复制代码

处理start_date <= rollback_date <= end_date 的数据，设置 end_date=9999-12-31

1. SELECT
2.     userid,
3.     mobile,
4.     regdate,
5.     start_date,
6.     '9999-12-31' AS end_date,
7.     '2' AS tag
8. FROM
9.     test.userhis
10. WHERE
11.     start_date <= '2020-06-22'
12.     AND end_date >= '2020-06-22';

复制代码

将前面两步的数据写入暂时表tmp（拉链表）

1. -- 删除暂时表DROP TABLE IF EXISTS test.tmp;-- 创建暂时表CREATE TABLE test.tmp ASSELECT     userid,     mobile,     regdate,     start_date,     end_date,     '1' AS tagFROM     test.userhisWHERE     end_date < '2020-06-22'UNION ALLSELECT
2.     userid,
3.     mobile,
4.     regdate,
5.     start_date,
6.     '9999-12-31' AS end_date,
7.     '2' AS tag
8. FROM
9.     test.userhis
10. WHERE
11.     start_date <= '2020-06-22'
12.     AND end_date >= '2020-06-22';
13. -- 查询结果并按照 userid 和 start_date 进行聚集SELECT * FROM test.tmp CLUSTER BY userid, start_date;

复制代码

模拟脚本：

1. zippertmp.sh

复制代码

写入的内容如下所示：

1. #!/bin/bash
3. # 加载环境变量
4. source /etc/profile
6. # 判断是否传递日期参数，如果没有则使用前一天的日期
7. if [ -n "$1" ]; then
8.   do_date=$1
9. else
10.   do_date=$(date -d "-1 day" +%F)
11. fi
13. # 定义SQL查询语句
14. sql="
15.   DROP TABLE IF EXISTS test.tmp;
17.   CREATE TABLE test.tmp AS
18.   SELECT userid, mobile, regdate, start_date, end_date, '1' AS tag
19.   FROM test.userhis
20.   WHERE end_date < '${do_date}'
21.   
22.   UNION ALL
23.   
24.   SELECT userid, mobile, regdate, start_date, '9999-12-31' AS end_date, '2' AS tag
25.   FROM test.userhis
26.   WHERE start_date <= '${do_date}'
27.   AND end_date >= '${do_date}';
28. "
30. # 执行Hive查询
31. hive -e "$sql"

复制代码

逐天回滚，检查数据

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。