Data Vault 初探（五） —— 定期裝載

說明：
1. 定期裝載的周期為每天一次。
2. 每天裝載自上次裝載后的變化數據
3. 建立源數據庫的過渡表用于CDC
4. 建立cdc_time表用于基于時間戳的CDC
5. 因為源庫上只有訂單銷售表有時間屬性，所以除了sales_order和sales_order_item拉取變化數據外，其它表都整體拉取到過渡區。實際環境中建議在源表設計上應該有created和last_updated兩個時間戳類型的字段。
6. 下表匯總了源庫各表的CDC方式

源數據庫表	過渡表	抽取模式
city	city_stg	整體、拉取
province	province_stg	整體、拉取
customer	customer_stg	整體、拉取
product_catagory	product_catagory_stg	整體、拉取
product	product_stg	整體、拉取
sales_order	sales_order_stg	CDC（每天）、拉取
sales_order_item	sales_order_item_stg	CDC（每天）、拉取

使用下面的腳本建立過渡區表和cdc_time表。

CREATE TABLE province_stg (province_id varchar(2),province_name varchar(20)
) ;CREATE TABLE product_catagory_stg (product_catagory_id varchar(2),product_catagory_name varchar(20)
) ;CREATE TABLE city_stg (city_id varchar(4),city_name varchar(20),province_id varchar(2)
) ;CREATE TABLE customer_stg (customer_id int(11),customer_name varchar(20),city_id varchar(4),cust_post_code varchar(6),cust_address varchar(50),ship_post_code varchar(6),ship_address varchar(50)
) ;CREATE TABLE product_stg (product_id int(11),product_name varchar(20),unit_price decimal(10,4),product_catagory_id varchar(2)
) ;CREATE TABLE sales_order_stg (sales_order_id int(11),order_time datetime,entry_time datetime,customer_id int(11),amount decimal(12,4),allocate_time datetime,packing_time datetime,ship_time datetime,receive_time datetime
) ;CREATE TABLE sales_order_item_stg (sales_order_item_id int(11),sales_order_id int(11),product_id int(11),unit_price decimal(10,4),quantity int(11)
) ;CREATE TABLE cdc_time
(last_load date,current_load date
);-- 插入數據倉庫開始日期
INSERT INTO cdc_time VALUES ('2015-03-01', '2015-03-01') ;
COMMIT ;

sales_order_item表是基于sales_order.sales_order_id更新的。為了避免對sales_order表的二次查詢，希望在對sales_order做CDC時同時獲取到sales_order_id用于后續的sales_order_item表的CDC。MySQL數據庫本身沒有提供類似于Oracle的returning這樣的語法，所以用內存表+觸發器的方式實現。使用下面的腳本建立內存表和sales_order_stg表上的insert觸發器。

USE dv;-- 建立內存表
CREATE TABLE sales_order_stg_insert ENGINE MEMORY 
SELECT sales_order_id, entry_time, allocate_time, packing_time, ship_time, receive_time 
FROM sales_order_stg WHERE FALSE;-- 建立insert觸發器
DELIMITER // DROP TRIGGER tr_sales_order_stg_insert_after //  
CREATE  TRIGGER tr_sales_order_stg_insert_after AFTER INSERT ON sales_order_stg   FOR EACH ROW BEGIN  INSERT INTO sales_order_stg_insert VALUES (NEW.sales_order_id, NEW.entry_time, NEW.allocate_time, NEW.packing_time, NEW.ship_time, NEW.receive_time);  END;  
//  DELIMITER ;

sales_order表的幾個時間字段可能修改，所以

新增訂單的條件是：
entry_time >= @last_load and entry_time < @current_load
修改訂單版本的條件是：
entry_time < @last_load and (allocate_time >= @last_load and allocate_time < @current_load
or packing_time >= @last_load and packing_time < @current_load
or ship_time >= @last_load and ship_time < @current_load
or receive_time >= @last_load and receive_time < @current_load)
sales_order_item表只基于sales_order.sales_order_id的新增訂單更新的。

使用下面的腳本進行每天定期裝載。

use dv;-- 設置附屬表的截止時間和生效時間
SET @pre_date = SUBDATE(CURRENT_DATE,1) ;
-- 設置CDC的上限時間
update cdc_time set current_load = current_date ;
select @last_load:=last_load,@current_load:=current_load from cdc_time;-- 裝載過渡表
truncate table customer_stg;
truncate table city_stg;
truncate table province_stg;
truncate table product_stg;
truncate table product_catagory_stg;
truncate table sales_order_stg;
truncate table sales_order_item_stg;
truncate table sales_order_stg_insert;insert into province_stg select * from province;
insert into city_stg select * from city;
insert into customer_stg select * from customer;
insert into product_catagory_stg select * from product_catagory;
insert into product_stg select * from product;
insert into sales_order_stg
select a.* from sales_order a
where entry_time >= @last_load and entry_time < @current_load
or (entry_time < @last_load and (allocate_time >= @last_load and allocate_time < @current_loador packing_time >= @last_load and packing_time < @current_loador ship_time >= @last_load and ship_time < @current_loador receive_time >= @last_load and receive_time < @current_load));
insert into sales_order_item_stg
select a.* from sales_order_item a, sales_order_stg_insert b
where a.sales_order_id = b.sales_order_idand b.entry_time >= @last_load and b.entry_time < @current_load;/*** 裝載中心表 ***/
insert into hub_product_catagory (product_catagory_id,record_source)  
select a.product_catagory_id,'source.product_catagory' 
from product_catagory_stg a left join hub_product_catagory b on a.product_catagory_id = b.product_catagory_id
where b.product_catagory_id is null;  insert into hub_customer (customer_id,record_source)  
select a.customer_id,'source.customer' 
from customer_stg a left join hub_customer b on a.customer_id = b.customer_id
where b.customer_id is null;  insert into hub_product (product_id,record_source)  
select a.product_id,'source.product' 
from product_stg a left join hub_product b on a.product_id = b.product_id
where b.product_id is null;  insert into hub_sales_order (sales_order_id,record_source)  
select sales_order_id,'source.sales_order' from sales_order_stg where entry_time >= @last_load and entry_time < @current_load; 
/*** 裝載中心表 ***//*** 裝載鏈接表 ***/
insert into link_order_customer (hub_sales_order_id,hub_customer_id,record_source)  
select hub_sales_order_id,hub_customer_id,'hub_sales_order,source.sales_order,hub_customer,source.customer'  
from hub_sales_order,sales_order_stg,hub_customer,customer_stg 
where hub_sales_order.sales_order_id = sales_order_stg.sales_order_id  and hub_customer.customer_id = customer_stg.customer_id  and sales_order_stg.customer_id = customer_stg.customer_idand sales_order_stg.entry_time >= @last_load and sales_order_stg.entry_time < @current_load;insert into link_order_product (hub_sales_order_id,hub_product_id,record_source)  
select hub_sales_order_id,hub_product_id,'hub_sales_order,hub_product,source.sales_order_item'  
from hub_sales_order,hub_product,sales_order_item_stg
where hub_sales_order.sales_order_id = sales_order_item_stg.sales_order_id  and hub_product.product_id = sales_order_item_stg.product_id;  insert into link_product_catagory (hub_product_id,hub_product_catagory_id,record_source)
select t1.hub_product_id hub_product_id, t1.hub_product_catagory_id hub_product_catagory_id,
'hub_product,product,hub_product_catagory' record_source
from 
(
select t2.hub_product_id hub_product_id, t3.hub_product_catagory_id hub_product_catagory_id
from product_stg t1, hub_product t2, hub_product_catagory t3
where t1.product_id = t2.product_id and t1.product_catagory_id = t3.product_catagory_id) t1
left join link_product_catagory t2 on t1.hub_product_id = t2.hub_product_id
where t2.hub_product_id is null;/*** 裝載鏈接表 ***//*** 裝載附屬表 ***/
-- 客戶附屬表
-- 修改老版本的截止日期
update hub_customer t1, sat_customer t2, 
(select customer_id,customer_name,city_name,province_name,cust_post_code,cust_address,ship_post_code,ship_address
from customer_stg t1,city_stg t2,province_stg t3
where t1.city_id = t2.city_id and t2.province_id = t3.province_id) t3
set t2.load_end_dts = @pre_date
where t1.hub_customer_id = t2.hub_customer_id
and t2.load_end_dts = '2200-01-01'
and t1.customer_id = t3.customer_id
and md5(concat(t2.customer_name,t2.city_name,t2.province_name,t2.cust_post_code,t2.cust_address,t2.ship_post_code,t2.ship_address))
<> 
md5(concat(t3.customer_name,t3.city_name,t3.province_name,t3.cust_post_code,t3.cust_address,t3.ship_post_code,t3.ship_address));-- 新增版本
insert into sat_customer   
(hub_customer_id,  
load_end_dts,  
record_source,  
customer_name,  
city_name,  
province_name,  
cust_post_code,  
cust_address,  
ship_post_code,  
ship_address) 
select 
t1.hub_customer_id,  
'2200-01-01',  
'hub_customer,customer,city,province', 
t3.customer_name,  
t3.city_name,  
t3.province_name,  
t3.cust_post_code,  
t3.cust_address,  
t3.ship_post_code,  
t3.ship_address 
from 
hub_customer t1, sat_customer t2, 
(select customer_id,customer_name,city_name,province_name,cust_post_code,cust_address,ship_post_code,ship_address
from customer_stg t1,city_stg t2,province_stg t3
where t1.city_id = t2.city_id and t2.province_id = t3.province_id) t3
where t1.hub_customer_id = t2.hub_customer_id
and t1.customer_id = t3.customer_id
and md5(concat(t2.customer_name,t2.city_name,t2.province_name,t2.cust_post_code,t2.cust_address,t2.ship_post_code,t2.ship_address))
<> 
md5(concat(t3.customer_name,t3.city_name,t3.province_name,t3.cust_post_code,t3.cust_address,t3.ship_post_code,t3.ship_address))
and exists (select 1 from sat_customer where hub_customer_id = t1.hub_customer_id 
and load_end_dts = @pre_date)
and not exists (select 1 from sat_customer where hub_customer_id = t1.hub_customer_id 
and load_end_dts = '2200-01-01');-- 新增記錄
insert into sat_customer   
(hub_customer_id,  
load_end_dts,  
record_source,  
customer_name,  
city_name,  
province_name,  
cust_post_code,  
cust_address,  
ship_post_code,  
ship_address)  
select  
t1.hub_customer_id,  
'2200-01-01',  
'hub_customer,customer,city,province',  
t2.customer_name,  
t2.city_name,  
t2.province_name,  
t2.cust_post_code,  
t2.cust_address,  
t2.ship_post_code,  
t2.ship_address 
from 
(select hub_customer.hub_customer_id, hub_customer.customer_id 
from hub_customer left join sat_customer 
on hub_customer.hub_customer_id = sat_customer.hub_customer_id
where sat_customer.hub_customer_id is null) t1,
(select customer_id,customer_name,city_name,province_name,cust_post_code,cust_address,ship_post_code,ship_address
from customer_stg t1,city_stg t2,province_stg t3
where t1.city_id = t2.city_id and t2.province_id = t3.province_id) t2
where t1.customer_id = t2.customer_id ; -- 訂單_產品附屬表
-- 新增記錄
insert into sat_order_product   
(link_order_product_id,  
load_end_dts,  
record_source,  
unit_price,  
quantity  
)  
select   
t1.link_order_product_id,  
'2200-01-01',  
'link_order_product,hub_sales_order,hub_product,sales_order_item',  
t4.unit_price,  
t4.quantity  
from link_order_product t1,hub_sales_order t2,hub_product t3,sales_order_item_stg t4 
where t1.hub_sales_order_id = t2.hub_sales_order_id  and t1.hub_product_id = t3.hub_product_id  and t4.sales_order_id = t2.sales_order_id  and t4.product_id = t3.product_id;-- 產品附屬表
-- 修改老版本的截止日期
update hub_product t1, sat_product t2, product_stg t3
set t2.load_end_dts = @pre_date
where t1.hub_product_id = t2.hub_product_id
and t2.load_end_dts = '2200-01-01'
and t1.product_id = t3.product_id
and md5(concat(t2.product_name,convert(t2.unit_price,char)))
<> 
md5(concat(t3.product_name,convert(t3.unit_price,char)));-- 新增版本
insert into sat_product 
(hub_product_id,  
load_end_dts,  
record_source,  
product_name,  
unit_price  
)  
select 
t1.hub_product_id,  
'2200-01-01',  
'hub_product,product', 
t3.product_name,  
t3.unit_price
from 
hub_product t1, sat_product t2, product_stg t3
where t1.hub_product_id = t2.hub_product_id
and t1.product_id = t3.product_id
and md5(concat(t2.product_name,convert(t2.unit_price,char)))
<> 
md5(concat(t3.product_name,convert(t3.unit_price,char)))
and exists (select 1 from sat_product where hub_product_id = t1.hub_product_id 
and load_end_dts = @pre_date)
and not exists (select 1 from sat_product where hub_product_id = t1.hub_product_id 
and load_end_dts = '2200-01-01');-- 新增記錄
insert into sat_product  
(hub_product_id,  
load_end_dts,  
record_source,  
product_name,  
unit_price  
)  
select  
t1.hub_product_id,  
'2200-01-01',  
'hub_product,product',  
t2.product_name,  
t2.unit_price
from 
(select hub_product.hub_product_id, hub_product.product_id 
from hub_product left join sat_product 
on hub_product.hub_product_id = sat_product.hub_product_id
where sat_product.hub_product_id is null) t1,
(
select product_id,product_name,unit_price 
from product_stg) t2
where t1.product_id = t2.product_id ; -- 產品類型附屬表
-- 修改老版本的截止日期
update hub_product_catagory t1, sat_product_catagory t2, product_catagory_stg t3
set t2.load_end_dts = @pre_date
where t1.hub_product_catagory_id = t2.hub_product_catagory_id
and t2.load_end_dts = '2200-01-01'
and t1.product_catagory_id = t3.product_catagory_id
and t2.product_catagory_name <> t3.product_catagory_name;-- 新增版本
insert into sat_product_catagory  
(hub_product_catagory_id,  
load_end_dts,  
record_source,  
product_catagory_name  
)   
select 
t1.hub_product_catagory_id,  
'2200-01-01',  
'hub_product_catagory,product_catagory',  
t3.product_catagory_name 
from 
hub_product_catagory t1, sat_product_catagory t2, product_catagory_stg t3
where t1.hub_product_catagory_id = t2.hub_product_catagory_id
and t1.product_catagory_id = t3.product_catagory_id
and t2.product_catagory_name <> t3.product_catagory_name
and exists (select 1 from sat_product_catagory where hub_product_catagory_id = t1.hub_product_catagory_id 
and load_end_dts = @pre_date)
and not exists (select 1 from sat_product_catagory where hub_product_catagory_id = t1.hub_product_catagory_id 
and load_end_dts = '2200-01-01');-- 新增記錄
insert into sat_product_catagory  
(hub_product_catagory_id,  
load_end_dts,  
record_source,  
product_catagory_name  
)  
select  
t1.hub_product_catagory_id,  
'2200-01-01',  
'hub_product_catagory,product_catagory', 
t2.product_catagory_name
from 
(select hub_product_catagory.hub_product_catagory_id, hub_product_catagory.product_catagory_id 
from hub_product_catagory left join sat_product_catagory 
on hub_product_catagory.hub_product_catagory_id = sat_product_catagory.hub_product_catagory_id
where sat_product_catagory.hub_product_catagory_id is null) t1,
(
select product_catagory_id,product_catagory_name
from product_catagory_stg) t2
where t1.product_catagory_id = t2.product_catagory_id ; -- 銷售訂單附屬表
-- 修改老版本的截止日期
update hub_sales_order t1, sat_sales_order t2, sales_order_stg t3
set t2.load_end_dts = @pre_date
where t1.hub_sales_order_id = t2.hub_sales_order_id
and t2.load_end_dts = '2200-01-01'
and t1.sales_order_id = t3.sales_order_id
and 
(t3.entry_time < @last_load and (t3.allocate_time >= @last_load and t3.allocate_time < @current_loador t3.packing_time >= @last_load and t3.packing_time < @current_loador t3.ship_time >= @last_load and t3.ship_time < @current_loador t3.receive_time >= @last_load and t3.receive_time < @current_load));-- 新增版本
insert into sat_sales_order  
(hub_sales_order_id,  
load_end_dts,  
record_source,  
order_time,  
entry_time,  
amount,  
allocate_time,  
packing_time,  
ship_time,  
receive_time  
) 
select 
t1.hub_sales_order_id,  
'2200-01-01',  
'hub_sales_order,sales_order',  
t3.order_time,  
t3.entry_time,  
t3.amount,  
t3.allocate_time,  
t3.packing_time,  
t3.ship_time,  
t3.receive_time 
from 
hub_sales_order t1, sat_sales_order t2, sales_order_stg t3
where t1.hub_sales_order_id = t2.hub_sales_order_id
and t1.sales_order_id = t3.sales_order_id
and 
(t3.entry_time < @last_load and (t3.allocate_time >= @last_load and t3.allocate_time < @current_loador t3.packing_time >= @last_load and t3.packing_time < @current_loador t3.ship_time >= @last_load and t3.ship_time < @current_loador t3.receive_time >= @last_load and t3.receive_time < @current_load))
and exists (select 1 from sat_sales_order where hub_sales_order_id = t1.hub_sales_order_id 
and load_end_dts = @pre_date)
and not exists (select 1 from sat_sales_order where hub_sales_order_id = t1.hub_sales_order_id 
and load_end_dts = '2200-01-01');-- 新增記錄
insert into sat_sales_order  
(hub_sales_order_id,  
load_end_dts,  
record_source,  
order_time,  
entry_time,  
amount,  
allocate_time,  
packing_time,  
ship_time,  
receive_time  
) 
select   
t1.hub_sales_order_id,  
'2200-01-01',  
'hub_sales_order,sales_order',  
t2.order_time,  
t2.entry_time,  
t2.amount,  
t2.allocate_time,  
t2.packing_time,  
t2.ship_time,  
t2.receive_time   
from hub_sales_order t1,sales_order_stg t2 
where t1.sales_order_id = t2.sales_order_idand t2.entry_time >= @last_load and t2.entry_time < @current_load;-- 更新時間戳表的last_load字段
update cdc_time set last_load = current_load ;commit ;