Kettle构建Hadoop ETL实践（八-2）：维度表技术

-- 在mysql的source库中建立源表  
use source;    
create table tree (c_child int, c_name varchar(100),c_parent int);    
create index idx1 on tree (c_parent);    
create unique index tree_pk on tree (c_child);    
-- 递归树结构，c_child是主键，c_parent是引用c_child的外键  
alter table tree add (constraint tree_pk primary key (c_child));    
alter table tree add (constraint tree_r01 foreign key (c_parent) references tree (c_child));    
-- 添加数据  
insert into tree (c_child, c_name, c_parent) 
values (1, '节点1', null),(2, '节点2', 1),(3, '节点3', 1),(4, '节点4', 1),    
       (5, '节点5', 2),(6, '节点6', 2),(7, '节点7', 2),(8, '节点8', 3),    
       (9, '节点9', 3),(10, '节点10', 4),(11, '节点11', 4); 
commit;  
  
-- 在hive的rds库中建立过渡表  
use rds;  
create table tree (c_child int,c_name string,c_parent int)
row format delimited fields terminated by ','
stored as textfile;
  
-- 在hive的dw库中建立相关维度表  
use dw;    
create table tree_dim    
(sk int,c_child int,c_name string ,c_parent int,  
  version int,effective_date date,expiry_date date)  
clustered by (sk) into 8 buckets      
stored as orc tblproperties ('transactional'='true');

--connect jdbc:mysql://node3:3306/source --delete-target-dir --password 123456 --table tree --target-dir /user/hive/warehouse/rds.db/tree --username root

use dw;    

truncate table tree_dim;  
insert into tree_dim    
select row_number() over (order by t1.c_child) + t2.sk_max, 
t1.c_child, t1.c_name, t1.c_parent, 1, '2020-03-01', '2200-01-01'    
from rds.tree t1 
cross join (select coalesce(max(sk),0) sk_max from tree_dim) t2;

use dw;    

update tree_dim set expiry_date = '${PRE_DATE}'        
 where tree_dim.sk in        
(select a.sk       
   from (select sk,c_child,c_name,c_parent      
           from tree_dim where expiry_date = '${MAX_DATE}') a 
   left join rds.tree b on a.c_child = b.c_child       
  where b.c_child is null 
     or (!(a.c_name <=> b.c_name) or !(a.c_parent <=> b.c_parent) ));       

insert into tree_dim      
select row_number() over (order by t1.c_child) + t2.sk_max,      
       t1.c_child, t1.c_name, t1.c_parent,
       t1.version, t1.effective_date, t1.expiry_date      
  from (select t2.c_child c_child, t2.c_name c_name, t2.c_parent c_parent,      
               t1.version + 1 version, '${PRE_DATE}' effective_date, '${MAX_DATE}' expiry_date        
          from tree_dim t1 
         inner join rds.tree t2 on t1.c_child = t2.c_child and t1.expiry_date = '${PRE_DATE}'        
          left join tree_dim t3 on t1.c_child = t3.c_child and t3.expiry_date = '${MAX_DATE}'       
         where (!(t1.c_name <=> t2.c_name) or !(t1.c_parent <=> t2.c_parent)) and t3.sk is null) t1        
 cross join (select coalesce(max(sk),0) sk_max from tree_dim) t2;      

insert into tree_dim      
select row_number() over (order by t1.c_child) + t2.sk_max,      
       t1.c_child, t1.c_name, t1.c_parent, 1, '${PRE_DATE}', '${MAX_DATE}'     
  from (select t1.* from rds.tree t1 
          left join tree_dim t2 on t1.c_child = t2.c_child where t2.sk is null) t1        
 cross join (select coalesce(max(sk),0) sk_max from tree_dim) t2; 

update rds.cdc_time set last_load=current_load;

-- 修改名称  
update tree set c_name = concat(c_name,'_1');

-- 修改名称
update tree set c_name = replace(c_name,'_1','_2')
where c_child in (1, 3, 5, 8, 11);    

-- 增加新的根节点，并改变原来的父子关系
insert into tree values (12, '节点12', null), (13, '节点13', 12);    
update tree 
set c_parent = (case when c_child = 1 then 12 else 13 end) 
where c_child in (1,3);

set foreign_key_checks=0;
update tree 
    set c_name = (case when c_child = 2 then '节点2_2' else '节点3_3' end) 
 where c_child in (2,3); 
delete from tree where c_child in (10,11,4); 

c_parent    c_child    distince
       1          1        0
      12          1        1
       1          2        1
       2          2        0
      12          2        2
       3          3        0
      12          3        2
      13          3        1
...

+---------+---------------+
| c_child | c_parent      |
+---------+---------------+
|       1 | 1, 12         |
|       2 | 2, 1, 12      |
|       3 | 3, 13, 12     |
|       5 | 5, 2, 1, 12   |
|       6 | 6, 2, 1, 12   |
|       7 | 7, 2, 1, 12   |
|       8 | 8, 3, 13, 12  |
|       9 | 9, 3, 13, 12  |
|      12 | 12            |
|      13 | 13, 12        |
+---------+---------------+

var level0="";
var level1="";
var level2="";
var level3="";
var strArray=new Array();
strArray=c_parent.split(", ");

var level0=strArray[0];
var level1=strArray[1];
var level2=strArray[2];
var level3=strArray[3];

12     1    2    5
12     1    2    6
12     1    2    7
12    13    3    8
12    13    3    9

select t0.c_0 c_0,t1.c_name c_0_name,t0.c_1 c_1,t2.c_name c_1_name,  
       t0.c_2 c_2,t3.c_name c_2_name,t0.c_3 c_3,t4.c_name c_3_name  
  from rds.tree_complanate t0   
 inner join (select * from rds.tree) t1 on t0.c_0= t1.c_child  
 inner join (select * from rds.tree) t2 on t0.c_1= t2.c_child  
 inner join (select * from rds.tree) t3 on t0.c_2= t3.c_child  
 inner join (select * from rds.tree) t4 on t0.c_3= t4.c_child;

use dw;

create table sales_order_fact_new
(order_number int COMMENT '销售订单号',                
 customer_sk int COMMENT '客户维度代理键',             
 product_sk int COMMENT '产品维度代理键',              
 order_date_sk int COMMENT '日期维度代理键',           
 order_amount decimal(10,2) COMMENT '销售金额',     
 order_quantity int COMMENT '销售数量',             
 request_delivery_date_sk int COMMENT '请求交付日期')
clustered by (order_number) into 8 buckets    
stored as orc tblproperties ('transactional'='true');

use dw; 
drop table 
drop table sales_order_fact;
alter table sales_order_fact_new rename to sales_order_fact;
drop table order_dim;

select a.order_number,
       c.customer_sk,
       d.product_sk,
       e.date_sk,
       a.order_amount,
       a.order_quantity,
       f.date_sk request_delivery_date_sk
  from rds.sales_order a,       
       dw.customer_dim c,
       dw.product_dim d,
       dw.date_dim e,
       dw.date_dim f
 where a.customer_number = c.customer_number
   and a.order_date >= c.effective_date
   and a.order_date < c.expiry_date
   and a.product_code = d.product_code
   and a.order_date >= d.effective_date
   and a.order_date < d.expiry_date
   and to_date(a.order_date) = e.dt
   and to_date(a.request_delivery_date) = f.dt
   and a.entry_date >= ? and a.entry_date < ?

use source;   
  
set @start_date := unix_timestamp('2020-10-31');    
set @end_date := unix_timestamp('2020-10-31 12:00:00');    
set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));    
set @amount := floor(1000 + rand() * 9000);   
set @quantity := floor(10 + rand() * 90);      
set @customer_number := floor(1 + rand() * 14);  
set @product_code := floor(1 + rand() * 5);  

insert into sales_order values (null,@customer_number,@product_code,@order_date,'2020-11-01',@order_date,@amount,@quantity);          
 
set @start_date := unix_timestamp('2020-10-31 12:00:01');    
set @end_date := unix_timestamp('2020-11-01');    
set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));    
set @amount := floor(1000 + rand() * 9000);   
set @quantity := floor(10 + rand() * 90);      
set @customer_number := floor(1 + rand() * 14);  
set @product_code := floor(1 + rand() * 5);  

insert into sales_order values (null,@customer_number,@product_code,@order_date,'2020-11-01',@order_date,@amount,@quantity);           
  
commit ;

use dw;    
-- 建立杂项维度表    
create table sales_order_attribute_dim (    
    sales_order_attribute_sk int comment '销售订单属性代理键',    
    verification_ind char(1) comment '是否审核，y or n',    
    credit_check_flag char(1) comment '是否检查过信用状态，y or n',    
    new_customer_ind char(1) comment '是否新用户，y or n',    
    web_order_flag char(1) comment '是否线上订单，y or n',    
    version int comment '版本号',    
    effective_date date comment '生效日期',    
    expiry_date date comment '到期日期'  
)  
clustered by (sales_order_attribute_sk) into 8 buckets      
stored as orc tblproperties ('transactional'='true');    
    
-- 生成杂项维度数据    
insert into sales_order_attribute_dim 
values (1, 'y', 'n', 'n', 'n', 1,'2020-03-01', '2200-01-01');  
...
-- 共插入16条记录
 
-- 建立杂项维度外键    
alter table sales_order_fact add columns (sales_order_attribute_sk int comment '订单属性代理键') ; 
  
-- 给源库的销售订单表增加对应的属性    
use source;     
alter table sales_order    
  add verification_ind char (1) after product_code    
, add credit_check_flag char (1) after verification_ind    
, add new_customer_ind char (1) after credit_check_flag    
, add web_order_flag char (1) after new_customer_ind ;    
  
-- 给销售订单过渡表增加对应的属性   
use rds;      
alter table sales_order add columns   
(  
verification_ind char(1) comment '是否审核，y or n',    
credit_check_flag char(1) comment '是否检查过信用状态，y or n',    
new_customer_ind char(1) comment '是否新用户，y or n',    
web_order_flag char(1) comment '是否线上订单，y or n'  
) ; 

select order_number,
       customer_number,
       product_code,
       order_date,
       entry_date,
       order_amount,
       order_quantity,
       request_delivery_date,
       verification_ind,
       credit_check_flag,
       new_customer_ind,
       web_order_flag
  from source.sales_order 
 where entry_date >= ? and entry_date < ?

select a.order_number,
       c.customer_sk,
       d.product_sk,
       e.date_sk,
       a.order_amount,
       a.order_quantity,
       f.date_sk request_delivery_date_sk,
       g.sales_order_attribute_sk
  from rds.sales_order a,       
       dw.customer_dim c,
       dw.product_dim d,
       dw.date_dim e,
       dw.date_dim f,
       dw.sales_order_attribute_dim g
 where a.customer_number = c.customer_number
   and a.order_date >= c.effective_date
   and a.order_date < c.expiry_date
   and a.product_code = d.product_code
   and a.order_date >= d.effective_date
   and a.order_date < d.expiry_date
   and to_date(a.order_date) = e.dt
   and to_date(a.request_delivery_date) = f.dt
   and a.verification_ind = g.verification_ind 
   and a.credit_check_flag = g.credit_check_flag    
   and a.new_customer_ind = g.new_customer_ind    
   and a.web_order_flag = g.web_order_flag
   and a.entry_date >= ? and a.entry_date < ?

use source;    
drop table if exists temp_sales_order_data;      
create table temp_sales_order_data as select * from sales_order where 1=0;   
  
set @start_date := unix_timestamp('2020-11-01');      
set @end_date := unix_timestamp('2020-11-02');      
  
set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));      
set @amount := floor(1000 + rand() * 9000);     
set @quantity := floor(10 + rand() * 90);        
set @customer_number := floor(1 + rand() * 14);  
set @product_code := floor(1 + rand() * 5);

insert into temp_sales_order_data 
values (1, @customer_number, @product_code, 'y', 'y', 'n', 'y', 
@order_date, '2020-11-04', @order_date, @amount, @quantity);  

... 
-- 一共添加各种属性组合的八条记录  

insert into sales_order            
select null,            
       customer_number,      
       product_code,  
       verification_ind,  
       credit_check_flag,  
       new_customer_ind,  
       web_order_flag,    
       order_date,      
       request_delivery_date,      
       entry_date,      
       order_amount,      
       order_quantity      
  from temp_sales_order_data t1 
 order by t1.order_date;    
commit; 

select concat(round(checked / (checked + not_checked) * 100),' % ')    
  from (select sum(case when credit_check_flag='y' then 1 else 0 end) checked,  
               sum(case when credit_check_flag='n' then 1 else 0 end) not_checked  
          from dw.sales_order_fact a, dw.sales_order_attribute_dim b    
         where new_customer_ind = 'y'                
          and a.sales_order_attribute_sk = b.sales_order_attribute_sk) t;

use dw;
create table zip_code_dim (    
    zip_code_sk int,    
    zip_code int,    
    city varchar(30),    
    state varchar(2),    
    version int,    
    effective_date date,    
    expiry_date date    
)  
clustered by (zip_code_sk) into 8 buckets          
stored as orc tblproperties ('transactional'='true');

insert into zip_code_dim
select row_number() over (order by t1.customer_zip_code), 
       customer_zip_code, 
       customer_city, 
       customer_state,
       1,'2020-03-01','2200-01-01' 
  from (select distinct customer_zip_code, customer_city, customer_state
          from customer_dim
         where customer_zip_code is not null
         union 
        select distinct shipping_zip_code, shipping_city, shipping_state 
          from customer_dim
         where shipping_zip_code is not null) t1;

create view customer_zip_code_dim 
(customer_zip_code_sk, customer_zip_code, customer_city, 
 customer_state, version, effective_date, expiry_date) as    
select zip_code_sk, zip_code, city, state, 
version,  effective_date, expiry_date    
      from  zip_code_dim;    
    
create view shipping_zip_code_dim 
(shipping_zip_code_sk, shipping_zip_code, shipping_city, 
 shipping_state, version, effective_date, expiry_date) as    
select zip_code_sk, zip_code, city, state, 
version, effective_date, expiry_date    
      from zip_code_dim; 

alter table sales_order_fact add columns 
(customer_zip_code_sk int comment '客户邮编代理键',
 shipping_zip_code_sk int comment '送货邮编代理键'
) ;

drop table if exists tmp;
create table tmp as select * from sales_order_fact;
truncate table sales_order_fact;

select t1.order_number,      
       t1.customer_sk,   
       t1.product_sk,    
       t1.order_date_sk,
       t1.order_amount,                               
       t1.order_quantity,
       t1.request_delivery_date_sk, 
       t1.sales_order_attribute_sk,    
       t2.customer_zip_code_sk,  
       t3.shipping_zip_code_sk   
  from dw.tmp t1   
  left join  
(select a.order_number order_number,c.customer_zip_code_sk customer_zip_code_sk  
   from dw.tmp a, dw.customer_dim b, dw.customer_zip_code_dim c  
  where a.customer_sk = b.customer_sk  
    and b.customer_zip_code = c.customer_zip_code) t2 
    on t1.order_number = t2.order_number  
  left join  
(select a.order_number order_number,c.shipping_zip_code_sk shipping_zip_code_sk  
   from dw.tmp a, dw.customer_dim b, dw.shipping_zip_code_dim c  
  where a.customer_sk = b.customer_sk  
    and b.shipping_zip_code = c.shipping_zip_code) t3 
    on t1.order_number = t3.order_number;

alter table customer_dim rename to customer_dim_old;  
create table customer_dim   
(customer_sk int comment '代理键',  
 customer_number int comment '客户编号，业务主键',  
 customer_name varchar(50) comment '客户名称',  
 customer_street_address varchar(50) comment '客户住址',  
 shipping_address varchar(50) comment '送货地址',  
 version int comment '版本号',  
 effective_date date comment '生效日期',  
 expiry_date date comment '到期日期')     
clustered by (customer_sk) into 8 buckets          
stored as orc tblproperties ('transactional'='true');  

insert into customer_dim  
select customer_sk,  
       customer_number,  
       customer_name,  
       customer_street_address,  
       shipping_address,  
       version,  
       effective_date,  
       expiry_date  
  from customer_dim_old;  
drop table customer_dim_old;  

-- 修改pa_customer_dim表，同样将邮编相关字段删除
...

select a.order_number, 
       b.customer_sk, 
       c.product_sk, 
       d.date_sk, 
       a.order_amount, 
       a.order_quantity,
       e.date_sk request_delivery_date_sk,
       f.sales_order_attribute_sk,
       g.customer_zip_code_sk,
       h.shipping_zip_code_sk
  from rds.sales_order a,
       dw.customer_dim b,
       dw.product_dim c,
       dw.date_dim d,
       dw.date_dim e,
       dw.sales_order_attribute_dim f,
       dw.customer_zip_code_dim g,
       dw.shipping_zip_code_dim h,
       rds.customer i 
 where a.customer_number = b.customer_number
   and a.order_date >= b.effective_date and a.order_date < b.expiry_date
   and a.product_code = c.product_code
   and a.order_date >= c.effective_date and a.order_date < c.expiry_date
   and to_date(a.order_date) = d.dt
   and to_date(a.request_delivery_date) = e.dt
   and a.verification_ind = f.verification_ind 
   and a.credit_check_flag = f.credit_check_flag    
   and a.new_customer_ind = f.new_customer_ind    
   and a.web_order_flag = f.web_order_flag 
   and a.customer_number = i.customer_number
   and i.customer_zip_code = g.customer_zip_code
   and a.order_date >= g.effective_date and a.order_date < g.expiry_date
   and i.shipping_zip_code = h.shipping_zip_code
   and a.order_date >= h.effective_date and a.order_date < h.expiry_date
   and a.entry_date >= ? and a.entry_date < ?

truncate table dw.pa_customer_dim;        
insert into dw.pa_customer_dim        
select distinct a.*              
  from dw.customer_dim a,  
       dw.sales_order_fact b,  
       dw.customer_zip_code_dim c     
 where c.customer_state = 'pa'   
   and b.customer_zip_code_sk = c.customer_zip_code_sk  
   and a.customer_sk = b.customer_sk;

use dw;    
create table annual_order_segment_dim (    
    segment_sk int,    
    segment_name varchar(30),    
    band_name varchar(50),    
    band_start_amount decimal(10,2),  
    band_end_amount decimal(10,2),  
    version int,  
    effective_date date,    
    expiry_date date    
)  
clustered by (segment_sk) into 8 buckets            
stored as orc tblproperties ('transactional'='true');   
  
insert into annual_order_segment_dim values  
(1, 'project', 'bottom', 0.01, 2500.00, 1, '2020-03-01', '2200-01-01'),
(2, 'project', 'low', 2500.01, 3000.00, 1, '2020-03-01', '2200-01-01'),
(3, 'project', 'mid-low', 3000.01, 4000.00, 1, '2020-03-01', '2200-01-01'),
(4, 'project', 'mid', 4000.01, 5500.00, 1, '2020-03-01', '2200-01-01'),
(5, 'project', 'mid_high', 5500.01, 6500.00, 1, '2020-03-01', '2200-01-01'),
(6, 'project', 'top', 6500.01, 99999999.99, 1, '2020-03-01', '2200-01-01'),
(7, 'grid', 'low', 0.01, 3000, 1, '2020-03-01', '2200-01-01'),
(8, 'grid', 'med', 3000.01, 6000.00, 1, '2020-03-01', '2200-01-01'),
(9, 'grid', 'high', 6000.01, 99999999.99, 1, '2020-03-01', '2200-01-01');
  
create table year_dim (    
    year_sk int,    
    year int) 
row format delimited fields terminated by ','  
stored as textfile;
    
create table annual_sales_order_fact (    
    customer_sk int,    
    year_sk int,    
    annual_order_amount decimal(10, 2))
row format delimited fields terminated by ','  
stored as textfile;
    
create table annual_customer_segment_fact (    
    segment_sk int,    
    customer_sk int,    
    year_sk int)
row format delimited fields terminated by ','  
stored as textfile;

select year from dw.order_date_dim

select a.customer_sk,   
       year_sk,   
       sum(order_amount) order_amount
  from dw.sales_order_fact a,  
       dw.year_dim c,  
       dw.order_date_dim d    
 where a.order_date_sk = d.order_date_sk    
   and c.year = d.year
   and d.year < 2021
 group by a.customer_sk, c.year_sk

select d.segment_sk,   
       a.customer_sk,   
       a.year_sk    
  from dw.annual_sales_order_fact a,   
       dw.annual_order_segment_dim d    
 where annual_order_amount >= band_start_amount    
   and annual_order_amount <= band_end_amount

select a.customer_sk csk,  
       a.year_sk ysk,  
       annual_order_amount amt,  
       segment_name sn,  
       band_name bn  
  from dw.annual_customer_segment_fact a,  
       dw.annual_order_segment_dim b,  
       dw.year_dim c,  
       dw.annual_sales_order_fact d  
 where a.segment_sk = b.segment_sk  
   and a.year_sk = c.year_sk  
   and a.customer_sk = d.customer_sk  
   and a.year_sk = d.year_sk  
cluster by csk, ysk, sn, bn;

select a.customer_sk,   
       year_sk,   
       sum(order_amount)    
  from dw.sales_order_fact a,  
       dw.year_dim c,  
       dw.order_date_dim d    
 where a.order_date_sk = d.order_date_sk    
   and c.year = d.year    
   and d.year = ${pre_year}   
 group by a.customer_sk, c.year_sk

select d.segment_sk,   
       a.customer_sk,   
       c.year_sk    
  from dw.annual_sales_order_fact a,   
       dw.year_dim c,  
       dw.annual_order_segment_dim d    
 where a.year_sk = c.year_sk  
   and c.year = ${pre_year}
   and annual_order_amount >= band_start_amount    
   and annual_order_amount <= band_end_amount