商品维度数据装载(拉链表)

拉链表设计：

1.采集当日全量数据存储到 ND（当日） 表中。
2.可从历史表中取出昨日全量数据存储到 OD（上日数据）表中。
3.用ND－OD为当日新增和变化的数据（即日增量数据）。
两个表进行全字段比较，将结果记录到tabel_I表中。
4.用OD－ND为状态到此结束需要封链的数据。 (需要修改END_DATE)；
两个表进行全字段比较，将结果记录到tabel_U表中 。
5.历史表（HIS）比ND表和OD表多两个字段（START_DATE，END_DATE）
6.将tabel_I表的内容全部insert插入到HIS表中。
START_DATE=‘当日’,END_DATE可设为’9999-12-31’ ；
7.更新封链记录的END_DATE。
历史表（HIS）和tabel_U表比较，START_DATE，END_DATE除外，以tabel_U表为准，两者交集将其END_DATE改成当日，说明该记录失效。
8.取数据时对日期进行条件选择即可,如：取20100101日的数据为
(where START_DATE<=‘20100101’ and END_DATE>‘20100101’ )。

拉链表的作用：

1. 数据量比较大。
2. 表中的部分字段会被更新，比如用户的地址，银行利率，订单的状态等。
3. 需要查看某一个时间点或者时间段的历史快照信息，比如，查看利率在历史某一个时间点的状态。
4. 变化的比例和频率不是很大，比如，总共有1000万的会员，每天新增和发生变化的有10万左右。
5. 如果对这边表每天都保留一份全量，那么每次全量中会保存很多不变的信息，对存储是极大的浪费;
   拉链历史表，既能满足反应数据的历史状态，又可以最大程度的节省存储。

使用拉链表解决商品SCD问题
5.1 dw层建表

-- dw层建表
DROP TABLE IF EXISTS `itcast_dw`.`dim_goods`;
CREATE TABLE `itcast_dw`.`dim_goods`(goodsId bigint,goodsSn string,productNo string,goodsName string,goodsImg string,shopId bigint,goodsType bigint,marketPrice double,shopPrice double,warnStock bigint,goodsStock bigint,goodsUnit string,goodsTips string,isSale bigint,isBest bigint,isHot bigint,isNew bigint,isRecom bigint,goodsCatIdPath string,goodsCatId bigint,shopCatId1 bigint,shopCatId2 bigint,brandId bigint,goodsDesc string,goodsStatus bigint,saleNum bigint,saleTime string,visitNum bigint,appraiseNum bigint,isSpec bigint,gallery string,goodsSeoKeywords string,illegalRemarks string,dataFlag bigint,createTime string,isFreeShipping bigint,goodsSerachKeywords string,modifyTime string,dw_start_date string,dw_end_date string
)
STORED AS PARQUET;

以上代码什么意思？
在dw层创建dim_goods表，并多添加两个字段一个是状态开始时间: dw_start_date string，一个是状态结束时间: dw_end_date string
他在干什么？
这个表用来保存表itcast_goods的历史数据和更新数据还有新增数据

5.2 具体步骤
拉链表设计一共分为以下几个步骤：
1、第一次全量导入
所有的ODS数据全部导入到拉链历史记录表中
2、增量导入（某天，举例：2018-09-09） 增量导入某天的数据到ODS分区 合并历史数据
通过连接查询方式更新
1全量导入
将所有 2019年09月08日以前创建的商品以及修改的数据全部导入到拉链历史记录表中
操作步骤：
1、使用Kettle将20190908以前的数据抽取到ods

SELECT *
FROM itcast_ods.itcast_goods
WHERE DATE_FORMAT(createtime, ‘%Y%m%d’) <= ‘20190908’ OR DATE_FORMAT(modifyTime, ‘%Y%m%d’) <= ‘20190908’;

在kettle的表输入组件里查找包含20190908及以前的日期数据
他在干什么？
把20190908日期及以前的数据作为全量数据输入到itcast_ods.itcast_goods

2、使用spark sql将全量数据导入到dw层维度表

set spark.sql.shuffle.partitions=1; --shuffle时的分区数，默认是200个
-- 使用spark sql将全量数据导入到dw层维度表
insert overwrite table `itcast_dw`.`dim_goods`
selectgoodsId,goodsSn,productNo,goodsName,goodsImg,shopId,goodsType,marketPrice,shopPrice,warnStock,goodsStock,goodsUnit,goodsTips,isSale,isBest,isHot,isNew,isRecom,goodsCatIdPath,goodsCatId,shopCatId1,shopCatId2,brandId,goodsDesc,goodsStatus,saleNum,saleTime,visitNum,appraiseNum,isSpec,gallery,goodsSeoKeywords,illegalRemarks,dataFlag,createTime,isFreeShipping,goodsSerachKeywords,modifyTime,case when modifyTime is not nullthen from_unixtime(unix_timestamp(modifyTime, 'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd')else from_unixtime(unix_timestamp(createTime, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') end as dw_start_date,'9999-12-31' as dw_end_date
from`itcast_ods`.`itcast_goods` t
where dt='20190908';

以上代码什么意思？
导入20190908分区全量数据到itcast_dw库的dim_goods表内
他在干什么？
把itcast_ods库的itcast_goods表把分区为20190908的数据作为全量数据导入到dw层的dim_goods表内，因为第一次导入多了两个日期字段，所以在导入进去时给每条数据的默认开始时间是每条数据的修改时间，结束时间定为9999-12-31以表示最新数据

2增量导入
将2019年09月09日创建的、修改的数据全部导入到历史拉链表中
操作步骤：
1、使用Kettle将20190909创建的、或者修改的数据抽取到ods

SELECT *
FROM itcast_goods
WHERE DATE_FORMAT(createtime, ‘%Y%m%d’) = ‘${dt}' OR DATE_FORMAT(modifyTime, '${dt}’;

以上代码什么意思？
在kettle的表输入组件里查找包含20190909及以前的日期数据
他在干什么？
==把20190909日期的数据作为新增数据输入itcast_ods.itcast_goods ==

2、编写spark-sql更新历史数据

-- 更新历史数据
selectdw.goodsId,dw.goodsSn,dw.productNo,dw.goodsName,dw.goodsImg,dw.shopId,dw.goodsType,dw.marketPrice,dw.shopPrice,dw.warnStock,dw.goodsStock,dw.goodsUnit,dw.goodsTips,dw.isSale,dw.isBest,dw.isHot,dw.isNew,dw.isRecom,dw.goodsCatIdPath,dw.goodsCatId,dw.shopCatId1,dw.shopCatId2,dw.brandId,dw.goodsDesc,dw.goodsStatus,dw.saleNum,dw.saleTime,dw.visitNum,dw.appraiseNum,dw.isSpec,dw.gallery,dw.goodsSeoKeywords,dw.illegalRemarks,dw.dataFlag,dw.createTime,dw.isFreeShipping,dw.goodsSerachKeywords,dw.modifyTime,dw.dw_start_date,case when dw.dw_end_date = '9999-12-31' and ods.goodsId is not nullthen '2019-09-08'else dw.dw_end_dateend as dw_end_date
from`itcast_dw`.`dim_goods` dwleft join (select * from `itcast_ods`.`itcast_goods` where dt='20190909') odson dw.goodsId = ods.goodsId;

以上代码什么意思？
更新dim_goods表中的历史数据
他在干什么？
用dim_goods表和itcast_goods表的20190909分区的数据进行left join，dim_goods表全部显示，itcast_goods表只要有关联的才显示，并且把相同goodsid的dim_goods表的结束日期为9999-12-31的数据进行修改，把结束时间改为2019-09-08

3、编写spark-sql获取当日数据

-- 今日数据
selectgoodsId,goodsSn,productNo,goodsName,goodsImg,shopId,goodsType,marketPrice,shopPrice,warnStock,goodsStock,goodsUnit,goodsTips,isSale,isBest,isHot,isNew,isRecom,goodsCatIdPath,goodsCatId,shopCatId1,shopCatId2,brandId,goodsDesc,goodsStatus,saleNum,saleTime,visitNum,appraiseNum,isSpec,gallery,goodsSeoKeywords,illegalRemarks,dataFlag,createTime,isFreeShipping,goodsSerachKeywords,modifyTime,case when modifyTime is not nullthen from_unixtime(unix_timestamp(modifyTime, 'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd')else from_unixtime(unix_timestamp(createTime, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') end as dw_start_date,'9999-12-31' as dw_end_date
from`itcast_ods`.`itcast_goods`
where dt = '20190909';

以上代码什么意思？
查看itcast_goods表内2019090分区的数据
他在干什么？
查看itcast_goods表内2019090分区的数据

4、将历史数据、当日数据合并加载到临时表

-- 将历史数据、当日数据合并加载到临时表
drop table if exists `itcast_dw`.`tmp_dim_goods_history`;
create table `itcast_dw`.`tmp_dim_goods_history`
as
selectdw.goodsId,dw.goodsSn,dw.productNo,dw.goodsName,dw.goodsImg,dw.shopId,dw.goodsType,dw.marketPrice,dw.shopPrice,dw.warnStock,dw.goodsStock,dw.goodsUnit,dw.goodsTips,dw.isSale,dw.isBest,dw.isHot,dw.isNew,dw.isRecom,dw.goodsCatIdPath,dw.goodsCatId,dw.shopCatId1,dw.shopCatId2,dw.brandId,dw.goodsDesc,dw.goodsStatus,dw.saleNum,dw.saleTime,dw.visitNum,dw.appraiseNum,dw.isSpec,dw.gallery,dw.goodsSeoKeywords,dw.illegalRemarks,dw.dataFlag,dw.createTime,dw.isFreeShipping,dw.goodsSerachKeywords,dw.modifyTime,dw.dw_start_date,case when dw.dw_end_date >= '2019-09-09' and ods.goodsId is not nullthen '2019-09-08'else dw.dw_end_dateend as dw_end_date
from`itcast_dw`.`dim_goods` dwleft join (select * from `itcast_ods`.`itcast_goods` where dt='20190909') odson dw.goodsId = ods.goodsId
union all
selectgoodsId,goodsSn,productNo,goodsName,goodsImg,shopId,goodsType,marketPrice,shopPrice,warnStock,goodsStock,goodsUnit,goodsTips,isSale,isBest,isHot,isNew,isRecom,goodsCatIdPath,goodsCatId,shopCatId1,shopCatId2,brandId,goodsDesc,goodsStatus,saleNum,saleTime,visitNum,appraiseNum,isSpec,gallery,goodsSeoKeywords,illegalRemarks,dataFlag,createTime,isFreeShipping,goodsSerachKeywords,modifyTime,case when modifyTime is not nullthen from_unixtime(unix_timestamp(modifyTime, 'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd')else from_unixtime(unix_timestamp(createTime, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') end as dw_start_date,'9999-12-31' as dw_end_date
from`itcast_ods`.`itcast_goods`
where dt = '20190909';

以上代码什么意思？
创建tmp_dim_goods_history临时表，并往表内添加数据
他在干什么？
创建tmp_dim_goods_history表用来存储历史数据和更新的数据及新增的数据，通过dim_goods表和itcast_goods表的20190909分区的数据left join更新历史数据的结束时间结果当成一个表，然后和itcast_goods分区20190909的数据进行合并，就能把历史数据更新，又能添加新增数据到这个临时表

5、将历史数据、当日数据导入到历史拉链表
– 将历史数据、当日数据导入到历史拉链表

insert overwrite table itcast_dw.dim_goods select * from itcast_dw.tmp_dim_goods_history;

以上代码什么意思？
把tmp_dim_goods_history临时表的数据全部插入到dim_goods最终表
他在干什么？
将tmp_dim_goods_history表存储的数据插入并覆盖到dim_goods表中

– 获取2019-09-09日的商品数据

select * from itcast_dw.dim_goods where dw_start_date <= ‘2019-09-09’ and dw_end_date >= ‘2019-09-09’ limit 10;

以上代码什么意思？
查找开始时间小于等于2019-09-09并且结束时间大于等于2019-09-09的十条数据
他在干什么？
查看数据的结果是否正确

3 测试
操作步骤：
1、将mysql中的一条数据的修改日期 改为 2019-09-10
2、设置kettle命名参数，重新抽取数据这一条数据到 20190910 分区
3、重新执行 spark-sql脚本加载数据到临时表

-- 导入2019-09-10的历史拉链数据
-- 将历史数据、当日数据合并加载到临时表
drop table if exists `itcast_dw`.`tmp_dim_goods_history`;
create table `itcast_dw`.`tmp_dim_goods_history`
as
selectdw.goodsId,dw.goodsSn,dw.productNo,dw.goodsName,dw.goodsImg,dw.shopId,dw.goodsType,dw.marketPrice,dw.shopPrice,dw.warnStock,dw.goodsStock,dw.goodsUnit,dw.goodsTips,dw.isSale,dw.isBest,dw.isHot,dw.isNew,dw.isRecom,dw.goodsCatIdPath,dw.goodsCatId,dw.shopCatId1,dw.shopCatId2,dw.brandId,dw.goodsDesc,dw.goodsStatus,dw.saleNum,dw.saleTime,dw.visitNum,dw.appraiseNum,dw.isSpec,dw.gallery,dw.goodsSeoKeywords,dw.illegalRemarks,dw.dataFlag,dw.createTime,dw.isFreeShipping,dw.goodsSerachKeywords,dw.modifyTime,dw.dw_start_date,case when dw.dw_end_date >= '2019-09-10' and ods.goodsId is not nullthen '2019-09-09'else dw.dw_end_dateend as dw_end_date
from`itcast_dw`.`dim_goods` dwleft join (select * from `itcast_ods`.`itcast_goods` where dt='20190910') odson dw.goodsId = ods.goodsId
union
selectgoodsId,goodsSn,productNo,goodsName,goodsImg,shopId,goodsType,marketPrice,shopPrice,warnStock,goodsStock,goodsUnit,goodsTips,isSale,isBest,isHot,isNew,isRecom,goodsCatIdPath,goodsCatId,shopCatId1,shopCatId2,brandId,goodsDesc,goodsStatus,saleNum,saleTime,visitNum,appraiseNum,isSpec,gallery,goodsSeoKeywords,illegalRemarks,dataFlag,createTime,isFreeShipping,goodsSerachKeywords,modifyTime,case when modifyTime is not nullthen from_unixtime(unix_timestamp(modifyTime, 'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd')else from_unixtime(unix_timestamp(createTime, 'yyyy-MM-dd HH:mm:ss'), 'yyyy-MM-dd') end as dw_start_date,'9999-12-31' as dw_end_date
from`itcast_ods`.`itcast_goods`
where dt = '20190910';

4、重新导入数据到历史拉链表
– 将历史数据、当日数据导入到历史拉链表

insert overwrite table itcast_dw.dim_goods
select * from itcast_dw.tmp_dim_goods_history;

5、查看对应商品id的历史拉链数据

select * from itcast_dw.dim_goods where goodsId = 100134;

将更新后的数据和更新之前的数据查出来
历史数据-----结束时间不是9999-12-31，应该是一个2019-09-10
最新的状态 -----束时间是9999-12-31