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

use source;  
alter table customer  
  add shipping_address varchar(50) after customer_state  
, add shipping_zip_code int after shipping_address  
, add shipping_city varchar(30) after shipping_zip_code  
, add shipping_state varchar(2) after shipping_city ;  
alter table sales_order add order_quantity int after order_amount ;

use rds;  
alter table customer add columns  
 (shipping_address varchar(50) comment '送货地址'  
, shipping_zip_code int comment '送货邮编'  
, shipping_city varchar(30) comment '送货城市'  
, shipping_state varchar(2) comment '送货省份') ;  
alter table sales_order add columns (order_quantity int comment '销售数量') ;

alter table customer_dim add columns  
 (shipping_address varchar(50) comment '送货地址'  
, shipping_zip_code int comment '送货邮编'  
, shipping_city varchar(30) comment '送货城市'  
, shipping_state varchar(2) comment '送货省份') ;  
alter table sales_order_fact add columns (order_quantity int comment '销售数量') ;

-- 装载customer维度  
-- 设置已删除记录和地址相关列上scd2的过期，用<=>运算符处理null值。
update customer_dim   
   set expiry_date = '${PRE_DATE}'   
 where customer_dim.customer_sk in    
(select a.customer_sk   
   from (select customer_sk,customer_number,customer_street_address, shipping_address
           from customer_dim where expiry_date = '${MAX_DATE}') a left join   
                rds.customer b on a.customer_number = b.customer_number   
          where b.customer_number is null 
             or (!(a.customer_street_address <=> b.customer_street_address) or !(a.shipping_address <=> b.shipping_address) ));

-- 处理地址列上scd2的新增行
insert into customer_dim  
select  
    row_number() over (order by t1.customer_number) + t2.sk_max,  
    t1.customer_number,  
    t1.customer_name,  
    t1.customer_street_address,  
    t1.customer_zip_code,  
    t1.customer_city,  
    t1.customer_state,
    t1.version,    
    t1.effective_date,    
    t1.expiry_date,        
    t1.shipping_address,
    t1.shipping_zip_code,
    t1.shipping_city,
    t1.shipping_state
from    
(    
select    
    t2.customer_number customer_number,  
    t2.customer_name customer_name,  
    t2.customer_street_address customer_street_address,  
    t2.customer_zip_code,  
    t2.customer_city,  
    t2.customer_state, 
    t1.version + 1 version,  
    '${PRE_DATE}' effective_date,    
    '${MAX_DATE}' expiry_date,
    t2.shipping_address shipping_address,  
    t2.shipping_zip_code shipping_zip_code,  
    t2.shipping_city shipping_city,  
    t2.shipping_state shipping_state
 from customer_dim t1   
inner join rds.customer t2    
   on t1.customer_number = t2.customer_number     
  and t1.expiry_date = '${PRE_DATE}'   
 left join customer_dim t3   
   on t1.customer_number = t3.customer_number   
  and t3.expiry_date = '${MAX_DATE}'    
where (!(t1.customer_street_address <=> t2.customer_street_address) or !(t1.shipping_address <=> t2.shipping_address) )
  and t3.customer_sk is null) t1    
cross join    
(select coalesce(max(customer_sk),0) sk_max from customer_dim) t2; 

-- 处理customer_name列上的scd1
drop table if exists tmp;  
create table tmp as  
select  
    a.customer_sk,  
    a.customer_number,  
    b.customer_name,  
    a.customer_street_address,  
    a.customer_zip_code,  
    a.customer_city,  
    a.customer_state,
    a.version,  
    a.effective_date,  
    a.expiry_date,
    a.shipping_address,
    a.shipping_zip_code,
    a.shipping_city,
    a.shipping_state
  from customer_dim a, rds.customer b    
 where a.customer_number = b.customer_number and !(a.customer_name <=> b.customer_name);    
delete from customer_dim where customer_dim.customer_sk in (select customer_sk from tmp);    
insert into customer_dim select * from tmp;

-- 处理新增的customer记录
insert into customer_dim  
select  
    row_number() over (order by t1.customer_number) + t2.sk_max,  
    t1.customer_number,  
    t1.customer_name,  
    t1.customer_street_address,  
    t1.customer_zip_code,  
    t1.customer_city,  
    t1.customer_state,
    1,  
    '${PRE_DATE}',
    '${MAX_DATE}',    
    t1.shipping_address,
    t1.shipping_zip_code,
    t1.shipping_city,
    t1.shipping_state
from    
(    
select t1.* from rds.customer t1 left join customer_dim t2 on t1.customer_number = t2.customer_number    
 where t2.customer_sk is null) t1    
cross join    
(select coalesce(max(customer_sk),0) sk_max from customer_dim) t2; 

select order_sk,
       customer_sk,
       product_sk,
       date_sk,
       order_amount,
       order_quantity
  from rds.sales_order a,
       dw.order_dim b,
       dw.customer_dim c,
       dw.product_dim d,
       dw.date_dim e
 where a.order_number = b.order_number
   and 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 a.entry_date >= ? and a.entry_date < ?

use source;    
update customer set shipping_address = customer_street_address, 
                    shipping_zip_code = customer_zip_code,
                    shipping_city = customer_city, 
                    shipping_state = customer_state ;

insert into customer
       (customer_name,
        customer_street_address,
        customer_zip_code,
        customer_city,
        customer_state,
        shipping_address,
        shipping_zip_code, 
        shipping_city,
        shipping_state)    
values ('online distributors',
        '2323 louise dr.', 
        17055,
        'pittsburgh',
        'pa',
        '2323 louise dr.',
        17055,
        'pittsburgh', 
        'pa') ;    
  
-- 新增订单日期为2020年10月25日的9条订单。  
set @start_date := unix_timestamp('2020-10-25');    
set @end_date := unix_timestamp('2020-10-26');  

drop table if exists temp_sales_order_data;    
create table temp_sales_order_data as select * from sales_order where 1=0;     

set @customer_number := floor(1 + rand() * 9);  
set @product_code := floor(1 + rand() * 4);       
set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));    
set @amount := floor(1000 + rand() * 9000);  
set @quantity := floor(10 + rand() * 90);  
insert into temp_sales_order_data 
values (121, @customer_number, @product_code, @order_date, @order_date, @amount, @quantity);    

... 新增9条订单 ...   
insert into sales_order    
select null,
       customer_number,
       product_code,
       order_date,
       entry_date,
       order_amount,
       order_quantity 
  from temp_sales_order_data 
 order by order_date;      
commit ; 

...
121    13    1    1029    2095.00    47
122    11    4    1029    5937.00    11
123    18    2    1029    2138.00    20
124    12    2    1029    2151.00    20
125    12    1    1029    5740.00    97
126    10    4    1029    4893.00    38
127    16    2    1029    2786.00    11
128    11    4    1029    6804.00    13
129    12    1    1029    6181.00    59

use dw;  
-- 建立月份维度表  
create table month_dim (    
    month_sk int comment '月份代理键',  
    month tinyint comment '月份',  
    month_name varchar(9) comment '月名称',  
    quarter tinyint comment '季度',  
    year smallint comment '年份'  
)  
comment '月份维度表'  
row format delimited fields terminated by ','  
stored as textfile;

select date_sk, dt, month, month_name, quarter, year from dw.date_dim;

[hdfs@manager~]$hdfs dfs -ls /user/hive/warehouse/dw.db/month_dim/
Found 1 items
-rw-r--r--   3 root hive       1326 2020-10-27 10:27 /user/hive/warehouse/dw.db/month_dim/month_dim.csv
[hdfs@manager~]$

hive> select * from month_dim;
OK
1    1    January      1    2018
2    2    February     1    2018
...
59    11    November     4    2022
60    12    December     4    2022
Time taken: 0.171 seconds, Fetched: 60 row(s)

use dw;    
create table pa_customer_dim (    
    customer_sk int comment '代理键',   
    customer_number int comment '客户编号，业务主键',    
    customer_name varchar(50) comment '客户名称',    
    customer_street_address varchar(50) comment '客户住址',  
    customer_zip_code int comment '邮编',    
    customer_city varchar(30) comment '所在城市',    
    customer_state varchar(2) comment '所在省份',   
    shipping_address varchar(50) comment '送货地址',   
    shipping_zip_code int comment '送货邮编',   
    shipping_city varchar(30) comment '送货城市',     
    shipping_state varchar(2) 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');

-- 装载customer维度
-- 设置已删除记录和地址相关列上scd2的过期
...
-- 处理地址列上scd2的新增行
...
-- 处理customer_name列上的scd1
...
-- 处理新增的customer记录
...
-- 重载pa客户维度
truncate table pa_customer_dim;    
insert into pa_customer_dim    
select customer_sk,customer_number,customer_name,
       customer_street_address,customer_zip_code,customer_city,customer_state,
       shipping_address,shipping_zip_code,shipping_city,shipping_state,
       version,effective_date,expiry_date    
  from customer_dim    
 where customer_state = 'pa' ;

FAILED: SemanticException [Error 10295]: INSERT OVERWRITE not allowed on table with OutputFormat that implements AcidOutputFormat while transaction manager that supports ACID is in use

use source;  

insert into customer  
(customer_name, customer_street_address, customer_zip_code, customer_city, customer_state, shipping_address, shipping_zip_code, shipping_city, shipping_state)  
values  
('pa customer', '1111 louise dr.', '17050', 'mechanicsburg', 'pa', '1111 louise dr.', '17050', 'mechanicsburg', 'pa'),  
('bigger customers', '7777 ridge rd.', '44102', 'cleveland', 'oh', '7777 ridge rd.', '44102', 'cleveland', 'oh'),   
('smaller stores', '8888 jennings fwy.', '44102', 'cleveland', 'oh', '8888 jennings fwy.', '44102', 'cleveland', 'oh'),  
('small-medium retailers', '9999 memphis ave.', '44102', 'cleveland', 'oh', '9999 memphis ave.', '44102', 'cleveland', 'oh'),  
('oh customer', '6666 ridge rd.', '44102', 'cleveland', 'oh', '6666 ridge rd.', '44102','cleveland', 'oh') ;  

commit;

select customer_number, customer_name, customer_state, effective_date, expiry_date 
  from dw.pa_customer_dim
 order by customer_number;

use dw;
-- 建立月份维度视图
create view month_dim as
select row_number() over (order by t1.year,t1.month) month_sk, t1.*
  from (select distinct month, month_name, quarter, year from date_dim) t1;

-- 建立PA维度视图
create view pa_customer_dim as 
select * from customer_dim where customer_state = 'pa';

-- in hive  
-- 修改数据仓库中的事实表sales_order_fact
use dw;   
alter table sales_order_fact add columns (request_delivery_date_sk int comment '请求交付日期') ;

-- 修改过渡区的sales_order表 
use rds;    
alter table sales_order add columns (request_delivery_date date comment '请求交付日期') ;    

-- in mysql
use source;    
alter table sales_order add request_delivery_date date after order_date ;

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

select b.order_sk,
       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.order_dim b,
       dw.customer_dim c,
       dw.product_dim d,
       dw.date_dim e,
       dw.date_dim f
 where a.order_number = b.order_number
   and 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;  
/*** 新增订单日期为2020年10月27日的3条订单。***/      
set @start_date := unix_timestamp('2020-10-27');      
set @end_date := unix_timestamp('2020-10-28');   
set @request_delivery_date := '2020-10-30';     
drop table if exists temp_sales_order_data;      
create table temp_sales_order_data as select * from sales_order where 1=0;       

set @customer_number := floor(1 + rand() * 14);  
set @product_code := floor(1 + rand() * 4);   
set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));      
set @amount := floor(1000 + rand() * 9000);    
set @quantity := floor(10 + rand() * 90);    
insert into temp_sales_order_data 
values (130, @customer_number, @product_code, @order_date, @request_delivery_date, @order_date, @amount, @quantity);      

... 插入3条订单记录 ...

insert into sales_order      
select null,customer_number,product_code,order_date,
request_delivery_date,entry_date,order_amount,order_quantity 
from temp_sales_order_data order by order_date;        
commit ;

use dw;  
select a.order_sk, request_delivery_date_sk, c.dt 
  from sales_order_fact a, date_dim b, date_dim c  
 where a.order_date_sk = b.date_sk   
   and a.request_delivery_date_sk = c.date_sk ; 

+-------------+---------------------------+-------------+
| a.order_sk  | request_delivery_date_sk  |    c.dt     |
+-------------+---------------------------+-------------+
| 130         | 1034                      | 2020-10-30  |
| 131         | 1034                      | 2020-10-30  |
| 132         | 1034                      | 2020-10-30  |
+-------------+---------------------------+-------------+
3 rows selected (36.268 seconds)

-- 使用表别名查询  
use dw;    
select order_date_dim.dt order_date,    
       request_delivery_date_dim.dt request_delivery_date,    
       sum(order_amount),count(*)    
  from sales_order_fact a,
       date_dim order_date_dim,    
       date_dim request_delivery_date_dim    
 where a.order_date_sk = order_date_dim.date_sk    
   and a.request_delivery_date_sk = request_delivery_date_dim.date_sk    
 group by order_date_dim.dt , request_delivery_date_dim.dt    
cluster by order_date_dim.dt , request_delivery_date_dim.dt;  
  
-- 使用视图查询  
use dw; 
 
-- 创建订单日期视图  
create view order_date_dim 
(order_date_sk, 
 order_date, 
 month, 
 month_name,  
 quarter, 
 year) 
as select * from date_dim;    

-- 创建请求交付日期视图
create view request_delivery_date_dim
(request_delivery_date_sk, 
 request_delivery_date, 
 month, 
 month_name, 
 quarter, 
 year)   
as select * from date_dim;  

-- 查询
select order_date,request_delivery_date,sum(order_amount),count(*)    
  from sales_order_fact a,order_date_dim b,request_delivery_date_dim c    
 where a.order_date_sk = b.order_date_sk    
   and a.request_delivery_date_sk = c.request_delivery_date_sk    
 group by order_date , request_delivery_date    
cluster by order_date , request_delivery_date;

select mid, money, name 
  from store distribute by mid sort by mid asc, money asc;

select mid, money, name from store cluster by mid sort by money;

create table date_dim (date_sk int, order_date date, delivery_date date);
create table sales_order_fact (date_sk int, order_amount int);

订单日期            请求交付日期
2020-10-27         2020-10-30
2020-10-28         2020-10-30
2020-10-29         2020-10-30
2020-10-27         2020-10-31
2020-10-28         2020-10-31
2020-10-29         2020-10-31

select product_category,year,quarter,month,order_amount
  from dw.sales_order_fact a, dw.product_dim b, dw.date_dim c    
 where a.product_sk = b.product_sk    
   and a.order_date_sk = c.date_sk

+-------------------+-------+----------+--------+----------------+
| product_category  | year  | quarter  | month  |  order_amount  |
+-------------------+-------+----------+--------+----------------+
| monitor           | 2020  |       4  |    10  |      42997.00  |
| peripheral        | 2020  |       4  |    10  |      37554.00  |
| storage           | 2020  |       1  |     3  |      98109.00  |
| storage           | 2020  |       2  |     4  |      51765.00  |
| storage           | 2020  |       2  |     5  |      89471.00  |
| storage           | 2020  |       2  |     6  |     143495.00  |
| storage           | 2020  |       3  |     7  |      87671.00  |
| storage           | 2020  |       3  |     8  |      58064.00  |
| storage           | 2020  |       3  |     9  |      10365.00  |
| storage           | 2020  |       4  |    10  |      91107.00  |
+-------------------+-------+----------+--------+----------------+

select product_category,year,quarter,month,dt,order_amount
  from dw.sales_order_fact a, dw.product_dim b, dw.date_dim c    
 where a.product_sk = b.product_sk    
   and a.order_date_sk = c.date_sk

var time_str;

if( sequence==1)
{
   time_str = "year: ".concat(time);
}
else if ( sequence==2)
{
   time_str = "quarter: ".concat(time);
}
else
{
   time_str = "month: ".concat(time);
}

+-------------------+-------------+---------------+
| product_category  |    time     | order_amount  |
+-------------------+-------------+---------------+
| monitor           | year: 2020  |     42997.00  |
| monitor           | quarter: 4  |     42997.00  |
| monitor           | month: 10   |     42997.00  |
| peripheral        | year: 2020  |     37554.00  |
| peripheral        | quarter: 4  |     37554.00  |
| peripheral        | month: 10   |     37554.00  |
| storage           | year: 2020  |    630047.00  |
| storage           | quarter: 1  |     98109.00  |
| storage           | month: 3    |     98109.00  |
| storage           | quarter: 2  |    284731.00  |
| storage           | month: 4    |     51765.00  |
| storage           | month: 5    |     89471.00  |
| storage           | month: 6    |    143495.00  |
| storage           | quarter: 3  |    156100.00  |
| storage           | month: 7    |     87671.00  |
| storage           | month: 8    |     58064.00  |
| storage           | month: 9    |     10365.00  |
| storage           | quarter: 4  |     91107.00  |
| storage           | month: 10   |     91107.00  |
+-------------------+-------------+---------------+

-- 增加促销期列  
use dw;
alter table month_dim add columns (campaign_session varchar(30) comment '促销期') ;

-- 建立促销期过渡表    
use rds;  
create table campaign_session 
(campaign_session varchar(30),month tinyint,year smallint)  
row format delimited fields terminated by ',' stored as textfile;

2020 First Campaign,1,2020
2020 First Campaign,2,2020
2020 First Campaign,3,2020
2020 First Campaign,4,2020
2020 Second Campaign,5,2020
2020 Second Campaign,6,2020
2020 Second Campaign,7,2020
2020 Third Campaign,8,2020
2020 Last Campaign,9,2020
2020 Last Campaign,10,2020
2020 Last Campaign,11,2020
2020 Last Campaign,12,2020

select a.month_sk,a.month,a.month_name,a.quarter,a.year,b.campaign_session 
  from dw.month_dim a left join rds.campaign_session b on a.year=b.year and a.month=b.month

,1,2020
2020 Early Spring Campaign,2,2020
2020 Early Spring Campaign,3,2020
,4,2020
2020 Spring Campaign,5,2020
,6,2020
2020 Last Campaign,7,2020
2020 Last Campaign,8,2020
,9,2020
,10,2020
,11,2020
,12,2020

select a.month_sk,a.month,a.month_name,a.quarter,a.year,
       if(length(trim(b.campaign_session))=0, a.month_name, b.campaign_session) campaign_session
  from dw.month_dim a left join rds.campaign_session b on a.year=b.year and a.month=b.month