大数据数仓之拉链表的设计思路

1. 为了解决大数据数据仓库中，不能更新的问题， 我们就采用拉链表进行弥补了这样的问题。拉链表是针对数据仓库设计中表存储数据的方式而定义的，顾名思义，所谓拉链，就是记录历史。记录一个事物从开始，一直到当前状态的所有变化的信息。

2.采用的架构

拉链表架构.png

3.实现过程

3.1 比如数据有一张表test_student

以下是2021年1月3号的数据，

id	name	create_time	update_time
1	hdfs	2021-02-03 09:45:16	2021-02-03 09:45:22
2	hive	2021-02-03 09:45:30	2021-02-03 09:45:34

3.2 将1月3号的数据通过sqoop 导入到hdfs 在加载到hive 中

将数据导入hdfs中

sqoop import \
--connect jdbc:mysql://slave03.com:10051/test_db \
--username root \
--password Mysql@20211231.20221 \
--target-dir hdfs://master01.com:8020/origin_data/db_hive/db/resource/test_db/2021-02-03 \
--delete-target-dir \
--query "SELECT id, NAME, create_time, update_time FROM test_student WHERE ( DATE_FORMAT(create_time, '%Y-%m-%d') = '2021-02-03' OR DATE_FORMAT(update_time, '%Y-%m-%d') = '2021-02-03' ) and  \$CONDITIONS" \
--num-mappers 1 \
--fields-terminated-by '\t' \
--compress \
--compression-codec lzop \
--null-string '\\N' \
--null-non-string '\\N'

3.3数据加载到hive

原始表：

CREATE external TABLE `ods_student` (
  `id` bigint COMMENT 'id',
  `name` string COMMENT '创建者',
  `create_time` TIMESTAMP COMMENT '创建时间',
  `update_time` TIMESTAMP COMMENT '更新时间'
)  COMMENT '原始表'
PARTITIONED BY (`dt` string)
row format delimited fields terminated by '\t'
STORED AS
  INPUTFORMAT 'com.hadoop.mapred.DeprecatedLzoTextInputFormat'
  OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
location '/warehouse/tablespace/external/hive/db_wudl/ods_student/';

3.4加载数据

load data inpath '/origin_data/db_hive/db/resource/test_db/2021-02-03' OVERWRITE into table db_wudl.ods_student partition(dt='2021-02-03')

查询如下：

在这里插入图片描述

3.4 初始化拉链表

create external table ods_student_his(
    `id` bigint COMMENT '用户id',
    `name` string COMMENT '姓名', 
    `create_time` TIMESTAMP COMMENT '创建时间',
    `update_time` TIMESTAMP COMMENT '操作时间',
    `start_date`  string COMMENT '有效开始日期',
    `end_date`  string COMMENT '有效结束日期'
) COMMENT '用户拉链表'
stored as parquet
location '/warehouse/tablespace/external/hive/db_wudl/ods_student_his/'
tblproperties ("parquet.compression"="lzo");

3.5 数据初始化拉链表

SET hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
insert overwrite table ods_student_his SELECT id, NAME, create_time, update_time, '2021-02-03', '9999-99-99' FROM ods_student oi WHERE oi.dt = '2021-02-03';

如果1月4号 在增加在增加一条数据并且在修改数据，如下：

1月三号的数据

id	name	create_time	update_time
1	hdfs	2021-02-03 09:45:16	2021-02-03 09:45:22
2	hive	2021-02-03 09:45:30	2021-02-03 09:45:34

1月四号的数据

在这里插入图片描述

这样就将修改的数据增加了一条flink 数据修改了之前的hive 数据

3.4再次同步2月4号的数据

sqoop import \
--connect jdbc:mysql://slave03.com:10051/test_db \
--username root \
--password Mysql@20211231.20221 \
--target-dir hdfs://master01.com:8020/origin_data/db_hive/db/resource/test_db/2021-02-04 \
--delete-target-dir \
--query "SELECT id, NAME, create_time, update_time FROM test_student WHERE ( DATE_FORMAT(create_time, '%Y-%m-%d') = '2021-02-04' OR DATE_FORMAT(update_time, '%Y-%m-%d') = '2021-02-04' ) and  \$CONDITIONS" \
--num-mappers 1 \
--fields-terminated-by '\t' \
--compress \
--compression-codec lzop \
--null-string '\\N' \
--null-non-string '\\N'

3.5 进行数据加载

load data inpath '/origin_data/db_hive/db/resource/test_db/2021-02-04' OVERWRITE into table db_wudl.ods_student partition(dt='2021-02-04')

3.6查询原始表

2021-2月4号数据.png

拉链临时表的创建

create external table ods_student_his_tmp(
    `id` bigint COMMENT '用户id',
    `name` string COMMENT '姓名', 
    `create_time` TIMESTAMP COMMENT '创建时间',
    `update_time` TIMESTAMP COMMENT '操作时间',
    `start_date`  string COMMENT '有效开始日期',
    `end_date`  string COMMENT '有效结束日期'
) COMMENT '订单拉链临时表'
stored as parquet
location '/warehouse/tablespace/external/hive/db_wudl/ods_student_his_tmp/'
tblproperties ("parquet.compression"="lzo");

3.7 原始表和拉链临时表进行关联

INSERT overwrite TABLE ods_student_his_tmp SELECT
    *
FROM
    (
        SELECT
            id,
            NAME,
            create_time,
            update_time,
            '2021-02-04' start_date,
            '9999-99-99' end_date
        FROM
            ods_student
        WHERE
            dt = '2021-02-04'
        UNION ALL
            SELECT
                uh.id,
                uh. NAME,
                uh.create_time,
                uh.update_time,
                uh.start_date,

            IF (
                ui.id IS NOT NULL
                AND uh.end_date = '9999-99-99',
                date_add(ui.dt ,- 1),
                uh.end_date
            ) end_date
            FROM
                ods_student_his uh
            LEFT JOIN (
                SELECT
                    *
                FROM
                    ods_student
                WHERE
                    dt = '2021-02-04'
            ) ui ON uh.id = ui.id
    ) his
ORDER BY
    his.id,
    start_date;

拉链临时表数据

在这里插入图片描述

覆盖

insert overwrite table ods_student_his  select * from ods_student_his_tmp;

最终显示：

hive拉链表最终结果.png