logo
搜索
我的图书馆

发文章

发文工具

其他工具

留言交流
搜索
分享

HAWQ取代传统数仓实践（十四）——事实表技术之累积快照

 不忘初心g2x5dp 2019-11-04

一、累积快照简介

        累积快照事实表用于定义业务过程开始、结束以及期间的可区分的里程碑事件。通常在此类事实表中针对过程中的关键步骤都包含日期外键，并包含每个步骤的度量，这些度量的产生一般都会滞后于数据行的创建时间。累积快照事实表中的一行，对应某一具体业务的多个状态。例如，当订单产生时会插入一行。当该订单的状态改变时，累积事实表行被访问并修改。这种对累积快照事实表行的一致性修改在三种类型的事实表（事务、周期快照、累积快照）中具有独特性，对于前面两类事实表只追加数据，不会对已经存在的行进行更新操作。除了日期外键与每个关键过程步骤关联外，累积快照事实表中还可以包含其它维度和可选退化维度的外键。
        累积快照事实表在库存、采购、销售、电商等业务领域都有广泛应用。比如在电商订单里面，下单的时候只有下单时间，但是在支付的时候，又会有支付时间，同理，还有发货时间，完成时间等。下面以销售订单数据仓库为例，讨论累积快照事实表的实现。
        假设希望跟踪以下五个销售订单的里程碑：下订单、分配库房、打包、配送和收货，分别用状态N、A、P、S、R表示。这五个里程碑的日期及其各自的数量来自源数据库的销售订单表。一个订单完整的生命周期由五行数据描述：下订单时生成一条销售订单记录；订单商品被分配到相应库房时，新增一条记录，存储分配时间和分配数量；产品打包时新增一条记录，存储打包时间和数量；类似的，订单配送和订单客户收货时也都分别新增一条记录，保存各自的时间戳与数量。为了简化示例，不考虑每种状态出现多条记录的情况（例如，一条订单中的产品可能是在不同时间点分多次出库），并且假设这五个里程碑是以严格的时间顺序正向进行的。
        对订单的每种状态新增记录只是处理这种场景的多种设计方案之一。如果里程碑的定义良好并且不会轻易改变，也可以考虑在源订单事务表中新增每种状态对应的数据列，例如，新增8列，保存每个状态的时间戳和数量。新增列的好处是仍然能够保证订单号的唯一性，并保持相对较少的记录数。但是，这种方案还需要额外增加一个last_modified字段记录订单的最后修改时间，用于Sqoop增量数据抽取。因为每条订单在状态变更时都会被更新，所以订单号字段已经不能作为变化数据捕获的比较依据。

二、建立累积快照表

1. 修改源库表结构

        执行下面的脚本将源数据库中销售订单事务表结构做相应改变，以处理五种不同的状态。
  1. use source;
  2. -- 修改销售订单事务表
  3. alter table sales_order
  4. change order_date status_date datetime,
  5. add order_status varchar(1) after status_date,
  6. change order_quantity quantity int;
  8. -- 删除sales_order表的主键
  9. alter table sales_order change order_number order_number int not null;
  10. alter table sales_order drop primary key;
  12. -- 建立新的主键
  13. alter table sales_order add id int unsigned not null auto_increment primary key comment '主键' first;
        说明：
  • 将order_date字段改名为status_date，因为日期不再单纯指订单日期，而是指变为某种状态日期。
  • 将order_quantity字段改名为quantity，因为数量变为某种状态对应的数量。
  • 在status_date字段后增加order_status字段，存储N、A、P、S、R等订单状态之一。它描述了status_date列对应的状态值，例如，如果一条记录的状态为N，则status_date列是下订单的日期。如果状态是R，status_date列是收货日期。
  • 每种状态都会有一条订单记录，这些记录具有相同的订单号，因此订单号不能再作为事务表的主键，需要删除order_number字段上的自增属性与主键约束。
  • 新增id字段作为销售订单表的主键，它是表中的第一个字段。

2. 重建销售订单外部表

        执行下面的语句重建销售订单外部表，使其与源表结构一致。
  1. set search_path=ext;  
  2. drop external table sales_order;   
  3. create external table sales_order      
  4. (  
  5.   id bigint,     
  6.   order_number int,          
  7.   customer_number int,          
  8.   product_code int,     
  9.   verification_ind char(1),  
  10.   credit_check_flag char(1),  
  11.   new_customer_ind char(1),  
  12.   web_order_flag char(1),   
  13.   status_date timestamp,
  14.   order_status char(1),  
  15.   request_delivery_date timestamp,    
  16.   entry_date timestamp,          
  17.   order_amount decimal(10 , 2 ),      
  18.   quantity int     
  19. )        
  20. location ('pxf://mycluster/data/ext/sales_order?profile=hdfstextsimple')        
  21.   format 'text' (delimiter=e',', null='null');       
  23. comment on table sales_order is '销售订单外部表'; 
  24. comment on column sales_order.id is '业务主键'; 
  25. comment on column sales_order.order_number is '订单号';      
  26. comment on column sales_order.customer_number is '客户编号';      
  27. comment on column sales_order.product_code is '产品编码';   
  28. comment on column sales_order.verification_ind is '审核标志';      
  29. comment on column sales_order.credit_check_flag is '信用检查标志';      
  30. comment on column sales_order.new_customer_ind is '客户首个订单标志';      
  31. comment on column sales_order.web_order_flag is '线上订单标志';      
  32. comment on column sales_order.status_date is '状态日期'; 
  33. comment on column sales_order.order_status is '订单状态';  
  34. comment on column sales_order.request_delivery_date is '请求交付日期';      
  35. comment on column sales_order.entry_date is '登记日期';      
  36. comment on column sales_order.order_amount is '销售金额';     
  37. comment on column sales_order.quantity is '数量';

3. 修改销售订单原始数据存储表

  1. set search_path=rds;
  2. alter table sales_order rename order_date to status_date;
  3. alter table sales_order rename order_quantity to quantity;
  4. alter table sales_order add column order_status char(1) default null;
  6. comment on column sales_order.status_date is '状态日期';
  7. comment on column sales_order.quantity is '数量';
  8. comment on column sales_order.order_status is '订单状态';
        说明：
  • 将销售订单事实表中order_date和order_quantity字段的名称修改为与源表一致。
  • 增加订单状态字段。
  • rds.sales_order并没有增加id列，原因有两个：一是该列只作为增量检查列，不用在原始数据表中存储；二是不需要再重新导入已有数据。

4. 修改销售订单事实表

  1. set search_path=tds;
  2. alter table sales_order_fact rename order_date_sk to status_date_sk;
  3. alter table sales_order_fact rename order_quantity to quantity;
  4. alter table sales_order_fact add column order_status char(1) default null;
  6. comment on column sales_order_fact.status_date_sk is '状态日期外键';
  7. comment on column sales_order_fact.quantity is '数量'; 
  8. comment on column sales_order_fact.order_status is '订单状态'; 
  10. create view v_sales_order_fact as
  11. select order_number,
  12.        customer_sk,
  13.        product_sk,
  14.        year_month,
  15.        order_amount,
  16.        request_delivery_date_sk,
  17.        sales_order_attribute_sk,
  18.        customer_zip_code_sk,
  19.        shipping_zip_code_sk,  
  20.        max(case order_status when 'N' then status_date_sk else null end) nd,
  21. 	   max(case order_status when 'N' then quantity else null end) nq,
  22. 	   max(case order_status when 'A' then status_date_sk else null end) ad,
  23. 	   max(case order_status when 'A' then quantity else null end) aq,
  24. 	   max(case order_status when 'P' then status_date_sk else null end) pd,
  25. 	   max(case order_status when 'P' then quantity else null end) pq,
  26. 	   max(case order_status when 'S' then status_date_sk else null end) sd,
  27. 	   max(case order_status when 'S' then quantity else null end) sq,
  28. 	   max(case order_status when 'R' then status_date_sk else null end) rd,
  29. 	   max(case order_status when 'R' then quantity else null end) rq
  30.   from sales_order_fact
  31.  group by order_number,
  32.           customer_sk,
  33.           product_sk,
  34.           year_month,
  35.           order_amount,
  36.           request_delivery_date_sk,
  37.           sales_order_attribute_sk,
  38.           customer_zip_code_sk,
  39.           shipping_zip_code_sk;
  41. -- 建立四个日期维度视图    
  42. create view v_allocate_date_dim 
  43. (allocate_date_sk, allocate_date, month, month_name, quarter, year)   
  44. as 
  45. select * from date_dim ;  
  47. create view v_packing_date_dim 
  48. (packing_date_sk, packing_date, month, month_name, quarter, year)   
  49. as    
  50. select * from date_dim ;  
  52. create view v_ship_date_dim 
  53. (ship_date_sk, ship_date, month, month_name, quarter, year)   
  54. as    
  55. select * from date_dim ;  
  57. create view v_receive_date_dim 
  58. (receive_date_sk, receive_date, month, month_name, quarter, year)   
  59. as    
  60. select * from date_dim ;
        说明：
  • 对销售订单事实表结构的修改与rds.sales_order类似。
  • 新建了一个视图v_sales_order_fact，将五个状态及其数量做行转列。
  • 建立四个日期角色扮演维度视图，用来获取相应状态的日期代理键。

三、重建增量抽取Sqoop作业

        使用下面的脚本重建Sqoop作业，因为源表会有多个相同的order_number，所以不能再用它作为检查字段，将检查字段改为id。
last_value=`sqoop job --show myjob_incremental_import | grep incremental.last.value | awk '{print $3}'` sqoop job --delete myjob_incremental_importsqoop job --create myjob_incremental_import -- import --connect 'jdbc:mysql://172.16.1.127:3306/source?usessl=false&user=dwtest&password=123456' --table sales_order --target-dir /data/ext/sales_order --compress --where 'entry_date < current_date()' --incremental append --check-column id --last-value $last_value

四、修改定期数据装载函数

  1. create or replace function fn_regular_load ()                
  2. returns void as                
  3. $$                
  4. declare                
  5.     -- 设置scd的生效时间              
  6.     v_cur_date date := current_date;                  
  7.     v_pre_date date := current_date - 1;              
  8.     v_last_load date;              
  9. begin              
  10.     -- 分析外部表              
  11.     analyze ext.customer;              
  12.     analyze ext.product;              
  13.     analyze ext.sales_order;              
  15.     -- 将外部表数据装载到原始数据表              
  16.     truncate table rds.customer;                
  17.     truncate table rds.product;               
  19.     insert into rds.customer select * from ext.customer;               
  20.     insert into rds.product select * from ext.product;              
  21.     insert into rds.sales_order         
  22.     select order_number,        
  23.            customer_number,        
  24.            product_code,        
  25.            status_date,        
  26.            entry_date,        
  27.            order_amount,        
  28.            quantity,        
  29.            request_delivery_date,    
  30.            verification_ind,    
  31.            credit_check_flag,    
  32.            new_customer_ind,    
  33.            web_order_flag,
  34.            order_status		   
  35.       from ext.sales_order;              
  37.     -- 分析rds模式的表              
  38.     analyze rds.customer;              
  39.     analyze rds.product;              
  40.     analyze rds.sales_order;              
  42.     -- 设置cdc的上限时间              
  43.     select last_load into v_last_load from rds.cdc_time;              
  44.     truncate table rds.cdc_time;              
  45.     insert into rds.cdc_time select v_last_load, v_cur_date;              
  47.     -- 装载客户维度              
  48.     insert into tds.customer_dim              
  49.     (customer_number,              
  50.      customer_name,              
  51.      customer_street_address,              
  52.      shipping_address,             
  53.      isdelete,              
  54.      version,              
  55.      effective_date)              
  56.     select case flag               
  57.                 when 'D' then a_customer_number              
  58.                 else b_customer_number              
  59.             end customer_number,              
  60.            case flag               
  61.                 when 'D' then a_customer_name              
  62.                 else b_customer_name              
  63.             end customer_name,              
  64.            case flag               
  65.                 when 'D' then a_customer_street_address              
  66.                 else b_customer_street_address              
  67.             end customer_street_address,              
  68.            case flag               
  69.                 when 'D' then a_shipping_address              
  70.                 else b_shipping_address              
  71.             end shipping_address,            
  72.            case flag               
  73.                 when 'D' then true              
  74.                 else false              
  75.             end isdelete,              
  76.            case flag               
  77.                 when 'D' then a_version              
  78.                 when 'I' then 1              
  79.                 else a_version + 1              
  80.             end v,              
  81.            v_pre_date              
  82.       from (select a.customer_number a_customer_number,              
  83.                    a.customer_name a_customer_name,              
  84.                    a.customer_street_address a_customer_street_address,              
  85.                    a.shipping_address a_shipping_address,              
  86.                    a.version a_version,              
  87.                    b.customer_number b_customer_number,              
  88.                    b.customer_name b_customer_name,              
  89.                    b.customer_street_address b_customer_street_address,              
  90.                    b.shipping_address b_shipping_address,              
  91.                    case when a.customer_number is null then 'I'              
  92.                         when b.customer_number is null then 'D'              
  93.                         else 'U'               
  94.                     end flag              
  95.               from v_customer_dim_latest a               
  96.               full join rds.customer b on a.customer_number = b.customer_number               
  97.              where a.customer_number is null -- 新增              
  98.                 or b.customer_number is null -- 删除              
  99.                 or (a.customer_number = b.customer_number               
  100.                     and not               
  101.                            (coalesce(a.customer_name,'') = coalesce(b.customer_name,'')               
  102.                         and coalesce(a.customer_street_address,'') = coalesce(b.customer_street_address,'')               
  103.                         and coalesce(a.shipping_address,'') = coalesce(b.shipping_address,'')               
  104.                         ))) t              
  105.              order by coalesce(a_customer_number, 999999999999), b_customer_number limit 999999999999;              
  107.     -- 装载产品维度              
  108.     insert into tds.product_dim              
  109.     (product_code,              
  110.      product_name,              
  111.      product_category,                   
  112.      isdelete,              
  113.      version,              
  114.      effective_date)              
  115.     select case flag               
  116.                 when 'D' then a_product_code              
  117.                 else b_product_code              
  118.             end product_code,              
  119.            case flag               
  120.                 when 'D' then a_product_name              
  121.                 else b_product_name              
  122.             end product_name,              
  123.            case flag               
  124.                 when 'D' then a_product_category              
  125.                 else b_product_category              
  126.             end product_category,              
  127.            case flag               
  128.                 when 'D' then true              
  129.                 else false              
  130.             end isdelete,              
  131.            case flag               
  132.                 when 'D' then a_version              
  133.                 when 'I' then 1              
  134.                 else a_version + 1              
  135.             end v,              
  136.            v_pre_date              
  137.       from (select a.product_code a_product_code,              
  138.                    a.product_name a_product_name,              
  139.                    a.product_category a_product_category,              
  140.                    a.version a_version,              
  141.                    b.product_code b_product_code,              
  142.                    b.product_name b_product_name,              
  143.                    b.product_category b_product_category,                             
  144.                    case when a.product_code is null then 'I'              
  145.                         when b.product_code is null then 'D'              
  146.                         else 'U'               
  147.                     end flag              
  148.               from v_product_dim_latest a               
  149.               full join rds.product b on a.product_code = b.product_code               
  150.              where a.product_code is null -- 新增              
  151.                 or b.product_code is null -- 删除              
  152.                 or (a.product_code = b.product_code               
  153.                     and not               
  154.                            (a.product_name = b.product_name               
  155.                         and a.product_category = b.product_category))) t              
  156.              order by coalesce(a_product_code, 999999999999), b_product_code limit 999999999999;              
  158.     -- 装载销售订单事实表                
  159.     insert into sales_order_fact                
  160.     select a.order_number,                
  161.            customer_sk,                
  162.            product_sk,     
  163.            e.date_sk,              
  164.            e.year * 100 + e.month,                   
  165.            order_amount,            
  166.            quantity,        
  167.            f.date_sk,    
  168.            g.sales_order_attribute_sk,  
  169.            h.customer_zip_code_sk,      
  170.            i.shipping_zip_code_sk,
  171.            a.order_status		   
  172.       from rds.sales_order a,               
  173.            v_customer_dim_his c,                
  174.            v_product_dim_his d,                
  175.            date_dim e,         
  176.            date_dim f,      
  177.            sales_order_attribute_dim g,   
  178.            v_customer_zip_code_dim h,      
  179.            v_shipping_zip_code_dim i,      
  180.            rds.customer j,  
  181.            rds.cdc_time k    
  182.      where a.customer_number = c.customer_number                
  183.        and a.status_date >= c.effective_date              
  184.        and a.status_date < c.expiry_date                 
  185.        and a.product_code = d.product_code                
  186.        and a.status_date >= d.effective_date              
  187.        and a.status_date < d.expiry_date                 
  188.        and date(a.status_date) = e.date          
  189.        and date(a.request_delivery_date) = f.date    
  190.        and a.verification_ind = g.verification_ind        
  191.        and a.credit_check_flag = g.credit_check_flag        
  192.        and a.new_customer_ind = g.new_customer_ind        
  193.        and a.web_order_flag = g.web_order_flag   
  194.        and a.customer_number = j.customer_number      
  195.        and j.customer_zip_code = h.customer_zip_code  
  196.        and j.shipping_zip_code = i.shipping_zip_code   
  197.        and a.entry_date >= k.last_load and a.entry_date < k.current_load;                            
  199.     -- 重载PA客户维度            
  200.     truncate table pa_customer_dim;              
  201.     insert into pa_customer_dim              
  202.     select distinct a.*                
  203.       from customer_dim a,    
  204.            sales_order_fact b,    
  205.            v_customer_zip_code_dim c       
  206.      where c.customer_state = 'pa'     
  207.        and b.customer_zip_code_sk = c.customer_zip_code_sk    
  208.        and a.customer_sk = b.customer_sk;    
  210.     -- 分析tds模式的表              
  211.     analyze customer_dim;              
  212.     analyze product_dim;              
  213.     analyze sales_order_fact;   
  214.     analyze pa_customer_dim;      
  216.     -- 更新时间戳表的last_load字段                
  217.     truncate table rds.cdc_time;              
  218.     insert into rds.cdc_time select v_cur_date, v_cur_date;              
  220. end;                
  221. $$                
  222. language plpgsql;
        需要修改定期数据装载中的相应列名。在装载事务事实表时，只用entry_date >= last_load and entry_date < current_load条件就可以过滤出所有新录入的、包括五种状态的订单，因为每种状态的订单都有自己对应的录入时间。
        HAWQ不能更新已有的表数据，因此在装载时只新增数据，然后通过视图转化为固定状态列的格式。注意，本示例中的累积周期快照视图仍然是以订单号字段作为逻辑上的主键。

五、测试

        在源数据库的销售订单事务表中新增两个销售订单记录。
  1. use source;
  3. set @order_date := from_unixtime(unix_timestamp('2017-06-02 00:00:01') + rand() * (unix_timestamp('2017-06-02 12:00:00') - unix_timestamp('2017-06-02 00:00:01')));
  4. set @request_delivery_date := from_unixtime(unix_timestamp(date_add(current_date, interval 5 day)) + rand() * 86400);
  5. set @amount := floor(1000 + rand() * 9000);
  6. set @quantity := floor(10 + rand() * 90);
  8. insert into source.sales_order values
  9. (null, 141, 1, 1, 'y', 'y', 'y', 'y', @order_date, 'N', @request_delivery_date,
  10. @order_date, @amount, @quantity);
  12. set @order_date := from_unixtime(unix_timestamp('2017-06-02 12:00:00') + rand() * (unix_timestamp('2017-06-03 00:00:00') - unix_timestamp('2017-06-02 12:00:00')));
  13. set @request_delivery_date := from_unixtime(unix_timestamp(date_add(current_date, interval 5 day)) + rand() * 86400);
  14. set @amount := floor(1000 + rand() * 9000);
  15. set @quantity := floor(10 + rand() * 90);
  17. insert into source.sales_order values
  18. (null, 142, 2, 2, 'y', 'y', 'y', 'y', @order_date, 'N', @request_delivery_date,
  19. @order_date, @amount, @quantity);
  21. commit;

        设置时间窗口。
  1. truncate table rds.cdc_time;              
  2. insert into rds.cdc_time select date '2017-06-02', date '2017-06-02';

        执行定期装载脚本。
~/regular_etl.sh

        查询v_sales_order_fact里的两个销售订单，确认定期装载成功。
  1. select a.order_number,  
  2.        c.order_date,  
  3.        d.allocate_date,  
  4.        e.packing_date,  
  5.        f.ship_date,  
  6.        g.receive_date  
  7.   from v_sales_order_fact a    
  8.   left join v_order_date_dim c on a.nd = c.order_date_sk  
  9.   left join v_allocate_date_dim d on a.ad = d.allocate_date_sk  
  10.   left join v_packing_date_dim e on a.pd = e.packing_date_sk  
  11.   left join v_ship_date_dim f on a.sd = f.ship_date_sk  
  12.   left join v_receive_date_dim g on a.rd = g.receive_date_sk  
  13.  where a.order_number > 140
  14.  order by order_number;
        查询结果如图1所示，只有order_date列有值，其它日期都是空，因为这两个订单是新增的，并且还没有分配库房、打包、配送或收货。
图1

        添加销售订单作为这两个订单的分配库房和/或打包的里程碑。
  1. use source;
  3. set @order_date := from_unixtime(unix_timestamp('2017-06-03 00:00:00') + rand() * (unix_timestamp('2017-06-03 12:00:00') - unix_timestamp('2017-06-03 00:00:00')));
  4. insert into sales_order
  5. select null,
  6. order_number,
  7. customer_number,
  8. product_code,
  9. verification_ind,
  10. credit_check_flag,
  11. new_customer_ind,
  12. web_order_flag,
  13. @order_date,
  14. 'A',
  15. request_delivery_date,
  16. @order_date,
  17. order_amount,
  18. quantity
  19. from sales_order
  20. where order_number = 141;
  23. set @order_date := from_unixtime(unix_timestamp('2017-06-03 12:00:00') + rand() * (unix_timestamp('2017-06-04 00:00:00') - unix_timestamp('2017-06-03 12:00:00')));
  24. insert into sales_order
  25. select null,
  26. order_number,
  27. customer_number,
  28. product_code,
  29. verification_ind,
  30. credit_check_flag,
  31. new_customer_ind,
  32. web_order_flag,
  33. @order_date,
  34. 'P',
  35. request_delivery_date,
  36. @order_date,
  37. order_amount,
  38. quantity
  39. from sales_order
  40. where id = 143;
  42. set @order_date := from_unixtime(unix_timestamp('2017-06-03 12:00:00') + rand() * (unix_timestamp('2017-06-04 00:00:00') - unix_timestamp('2017-06-03 12:00:00')));
  43. insert into sales_order
  44. select null,
  45. order_number,
  46. customer_number,
  47. product_code,
  48. verification_ind,
  49. credit_check_flag,
  50. new_customer_ind,
  51. web_order_flag,
  52. @order_date,
  53. 'A',
  54. request_delivery_date,
  55. @order_date,
  56. order_amount,
  57. quantity
  58. from sales_order
  59. where order_number = 142;
  61. commit;
        设置时间窗口。
  1. truncate table rds.cdc_time;              
  2. insert into rds.cdc_time select date '2017-06-03', date '2017-06-03';
        执行定期装载脚本。
~/regular_etl.sh
        查询v_sales_order_fact表里的两个销售订单，确认定期装载成功。  查询结果如图2所示。第一个订单具有了allocate_date和packing_date，第二个只具有allocate_date。
图2

        添加销售订单作为这两个订单后面的里程碑：打包、配送和/或收货。注意四个日期可能相同。
  1. use source;    
  3. set @order_date := from_unixtime(unix_timestamp('2017-06-04 00:00:00') + rand() * (unix_timestamp('2017-06-04 12:00:00') - unix_timestamp('2017-06-04 00:00:00'))); 
  4. insert into sales_order 
  5. select null,
  6.        order_number,
  7.        customer_number,
  8.        product_code,
  9.        verification_ind,
  10.        credit_check_flag,
  11.        new_customer_ind,
  12.        web_order_flag,
  13.        @order_date,
  14.        'S',
  15.        request_delivery_date,
  16.        @order_date,
  17.        order_amount,
  18.        quantity      
  19.   from sales_order 
  20.  where order_number = 141
  21.  order by id desc
  22.  limit 1;
  24. set @order_date := from_unixtime(unix_timestamp('2017-06-04 12:00:00') + rand() * (unix_timestamp('2017-06-05 00:00:00') - unix_timestamp('2017-06-04 12:00:00'))); 
  25. insert into sales_order 
  26. select null,
  27.        order_number,
  28.        customer_number,
  29.        product_code,
  30.        verification_ind,
  31.        credit_check_flag,
  32.        new_customer_ind,
  33.        web_order_flag,
  34.        @order_date,
  35.        'R',
  36.        request_delivery_date,
  37.        @order_date,
  38.        order_amount,
  39.        quantity      
  40.   from sales_order 
  41.  where order_number = 141
  42.  order by id desc
  43.  limit 1;
  45. set @order_date := from_unixtime(unix_timestamp('2017-06-04 12:00:00') + rand() * (unix_timestamp('2017-06-05 00:00:00') - unix_timestamp('2017-06-04 12:00:00'))); 
  46. insert into sales_order 
  47. select null,
  48.        order_number,
  49.        customer_number,
  50.        product_code,
  51.        verification_ind,
  52.        credit_check_flag,
  53.        new_customer_ind,
  54.        web_order_flag,
  55.        @order_date,
  56.        'P',
  57.        request_delivery_date,
  58.        @order_date,
  59.        order_amount,
  60.        quantity      
  61.   from sales_order 
  62.  where order_number = 142
  63.  order by id desc
  64.  limit 1;
  66. commit;
        设置时间窗口。
  1. truncate table rds.cdc_time;
  2. insert into rds.cdc_time select date '2017-06-04', date '2017-06-04';
        执行定期装载脚本。
~/regular_etl.sh

        查询v_sales_order_fact表里的两个销售订单，确认定期装载成功。查询结果如图3所示。第一个订单号为141的订单，具有了全部日期，这意味着订单已完成（客户已经收货）。第二个订单已经打包，但是还没有配送。


图3

六、修改周期快照表装载函数

        累积快照将原来的一个数量order_quantity变为了每种状态对应一个数量，因此需要修改周期快照表装载函数fn_month_sum。该函数汇总月底订单金额和数量，我们必须重新定义数量，假设需要统计的是新增订单中的数量。修改后的函数如下。
  1. create or replace function tds.fn_month_sum(p_year_month int)   
  2. returns void as   
  3. $$  
  4. declare      
  5.     sqlstring varchar(1000);     
  6. begin  
  7.     -- 幂等操作，先删除上月数据  
  8.     sqlstring := 'truncate table month_end_sales_order_fact_1_prt_p' || cast(p_year_month as varchar);  
  9.     execute sqlstring;  
  11.     -- 插入上月销售汇总数据  
  12.     insert into month_end_sales_order_fact    
  13.     select t1.year_month,   
  14.            t2.product_sk,   
  15.            coalesce(t2.month_order_amount,0),   
  16.            coalesce(t2.month_order_quantity,0)   
  17.       from (select p_year_month year_month) t1   
  18.       left join (select year_month, product_sk, sum(order_amount) month_order_amount, sum(quantity) month_order_quantity  
  19.                    from sales_order_fact   
  20.                   where year_month = p_year_month and coalesce(order_status,'N') = 'N' 
  21.                   group by year_month,product_sk) t2   
  22.            on t1.year_month = t2.year_month;  
  24. end;  
  25. $$      
  26. language plpgsql;

    本站是提供个人知识管理的网络存储空间，所有内容均由用户发布，不代表本站观点。请注意甄别内容中的联系方式、诱导购买等信息，谨防诈骗。如发现有害或侵权内容，请点击一键举报。
    转藏 分享 献花（0）

    来自： 不忘初心g2x5dp > 《大数据》

    举报/认领

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多