基于Hadoop生态圈的数据仓库实践 —— 进阶技术(十一)

1、基于Hadoop生态圈的数据仓库实践 进阶技术（十一）十一、多重星型模式 从“进阶技术”开始，已经通过增加列和表扩展了数据仓库，在进阶技术（五） “快照”里增加了第二个事实表，month_end_sales_order_fact表。这之后数据仓库模式就有了两个事实表（第一个是在开始建立数据仓库时创建的sales_order_fact表）。有了这两个事实表的数据仓库就是一个标准的双星型模式。本节将在现有的维度数据仓库上再增加一个新的星型结构。与现有的与销售关联的星型结构不同，新的星型结构关注的是产品业务领域。新的星型结构有一个事实表和一个维度表，用于存储数据仓库中的产品数据。1. 一个新的星型模

2、式 下图显示了扩展后的数据仓库模式。模式中有三个星型结构。sales_order_fact表是第一个星型结构的事实表，与其相关的维度表是customer_dim、product_dim、date_dim和sales_order_attribute_dim表。month_end_sales_order_fact表是第二个星型结构的事实表。product_dim和month_dim是其对应的维度表。第一个和第二个星型结构共享product_dim维度表。第二个星型结构的事实表和月份维度数据分别来自于第一个星型结构的事实表和date_dim维度表。它们不从源数据获得数据。第三个星型模式的事实表是新建

3、的production_fact表。它的维度除了存储在已有的date_dim和product_dim表，还有一个新的factory_dim表。第三个星型结构的数据来自源数据。 执行下面的脚本建立第三个星型模式中的新表和对应的源数据表。sql view plain copy 在CODE上查看代码片派生到我的代码片- 在MySQL源库上建立工厂表和每日产品表 USE source; CREATE TABLE factory_master ( factory_code INT, factory_name CHAR(30), factory_street_address CHAR(50), facto

4、ry_zip_code INT(5), factory_city CHAR(30), factory_state CHAR(2) ); alter table factory_master add primary key (factory_code); CREATE TABLE daily_production ( product_code INT, production_date DATE, factory_code INT, production_quantity INT ); ALTER TABLE daily_production ADD FOREIGN KEY (factory_co

5、de) REFERENCES factory_master(factory_code) ON DELETE CASCADE ON UPDATE CASCADE ; ALTER TABLE daily_production ADD FOREIGN KEY (product_code) REFERENCES product(product_code) ON DELETE CASCADE ON UPDATE CASCADE ; alter table daily_production add primary key (product_code,production_date,factory_code

6、); - 在Hive的rds库上建立相应的过渡表 USE rds; CREATE TABLE factory_master ( factory_code INT, factory_name VARCHAR(30), factory_street_address VARCHAR(50), factory_zip_code INT, factory_city VARCHAR(30), factory_state VARCHAR(2) ); CREATE TABLE daily_production ( product_code INT, production_date DATE, factory_

7、code INT, production_quantity INT ); - 在Hive的dw库上建立相应的维度表和事实表 USE dw; CREATE TABLE factory_dim ( factory_sk INT, factory_code INT, factory_name VARCHAR(30), factory_street_address VARCHAR(50), factory_zip_code INT, factory_city VARCHAR(30), factory_state VARCHAR(2), version int, effective_date DATE,

8、 expiry_date DATE ) clustered by (factory_sk) into 8 buckets stored as orc tblproperties ('transactional'='true'); CREATE TABLE production_fact ( product_sk INT , production_date_sk INT , factory_sk INT , production_quantity INT ); 2. 新建定期导入脚本文件（1）新建抽取作业plain view plain copy 在CODE上查看

9、代码片派生到我的代码片# 建立增量抽取每日产品表的作业，以production_date作为检查列，初始值是'1900-01-01' last_value='1900-01-01' sqoop job -delete myjob_incremental_import_daily_production -meta-connect jdbc:hsqldb:hsql:/cdh2:16000/sqoop sqoop job -meta-connect jdbc:hsqldb:hsql:/cdh2:16000/sqoop -create myjob_incremental

10、_import_daily_production - import -connect "jdbc:mysql:/cdh1:3306/source?useSSL=false&user=root&password=mypassword" -table daily_production -columns "product_code,production_date,factory_code,production_quantity" -hive-import -hive-table rds.daily_production -incremental

11、 append -check-column production_date -last-value $last_value 新建定期装载每日产品脚本文件regular_etl_daily_production.sh，内容如下。plain view plain copy 在CODE上查看代码片派生到我的代码片#!/bin/bash # 全量抽取工厂表 sqoop import -connect jdbc:mysql:/cdh1:3306/source?useSSL=false -username root -password mypassword -table factory_master -h

12、ive-impo rt -hive-table rds.factory_master -hive-overwrite # 增量抽取每日产品表 sqoop job -exec myjob_incremental_import_daily_production -meta-connect jdbc:hsqldb:hsql:/cdh2:16000/sqoop # 调用regular_etl_daily_production.sql文件执行定期装载 beeline -u jdbc:hive2:/cdh2:10000/dw -f regular_etl_daily_production.sql 为了和其

13、它定期装载脚本共用环境和时间窗口设置，新建一个set_time.sql脚本，内容如下。sql view plain copy 在CODE上查看代码片派生到我的代码片- 设置变量以支持事务 set hive.support.concurrency=true; set hive.exec.dynamic.partition.mode=nonstrict; set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager; set pactor.initiator.on=true; set pactor.worker.thread

14、s=1; USE dw; - 设置SCD的生效时间和过期时间 SET hivevar:cur_date = CURRENT_DATE(); SET hivevar:pre_date = DATE_ADD($hivevar:cur_date,-1); SET hivevar:max_date = CAST('2200-01-01' AS DATE); - 设置CDC的上限时间 INSERT OVERWRITE TABLE rds.cdc_time SELECT last_load, $hivevar:cur_date FROM rds.cdc_time; 新建regular_et

15、l_daily_production.sql脚本文件内容如下。sql view plain copy 在CODE上查看代码片派生到我的代码片- 设置环境与时间窗口 !run /root/set_time.sql - 工厂信息很少修改，一般不需要保留历史，所以使用SCD1 drop table if exists tmp; create table tmp as select a.factory_sk, a.factory_code, b.factory_name, b.factory_street_address, b.factory_zip_code, b.factory_city, b.f

16、actory_state, a.version, a.effective_date, a.expiry_date from factory_dim a,rds.factory_master b where a.factory_code = b.factory_code and !(a.factory_name <=> b.factory_name and a.factory_street_address <=> b.factory_street_address and a.factory_zip_code <=> b.factory_zip_code and

17、 a.factory_city <=> b.factory_city and a.factory_state <=> b.factory_state); delete from factory_dim where factory_dim.factory_sk in (select factory_sk from tmp); insert into factory_dim select * from tmp; - 添加新的工厂信息 INSERT INTO factory_dim SELECT ROW_NUMBER() OVER (ORDER BY t1.factory_c

18、ode) + t2.sk_max, t1.factory_code, t1.factory_name, t1.factory_street_address, t1.factory_zip_code, t1.factory_city, t1.factory_state, 1, $hivevar:pre_date, $hivevar:max_date FROM ( SELECT t1.* FROM rds.factory_master t1 LEFT JOIN factory_dim t2 ON t1.factory_code = ory_code WHERE t2.factory_sk IS N

19、ULL) t1 CROSS JOIN (SELECT COALESCE(MAX(factory_sk),0) sk_max FROM factory_dim) t2; - 装载每日产品事实表 INSERT INTO production_fact SELECT duct_sk , c.date_sk , d.factory_sk , production_quantity FROM rds.daily_production a , product_dim b , date_dim c , factory_dim d WHERE production_date = $hivevar:p

20、re_date AND duct_code = duct_code AND duction_date >= b.effective_date AND duction_date <= b.expiry_date AND duction_date = c.date AND a.factory_code = d.factory_code ; 3. 测试 到目前为止已经讨论了第三个星型结构里的所有表，现在做一些测试。首先需要一些工厂信息。执行下面的语句向源数据库的factory_master表中装载四个工厂信息。sql view plain

21、 copy 在CODE上查看代码片派生到我的代码片USE source; INSERT INTO factory_master VALUES ( 1, 'First Ftory', '11111 Lichtman St.', 17050, 'Mechanicsburg', 'PA' ) , ( 2, 'Second Factory', '22222 Stobosky Ave.', 17055, 'Pittsburgh', 'PA' ) , ( 3, 'Thir

22、d Factory', '33333 Fritze Rd.', 17050, 'Mechanicsburg', 'PA' ) , ( 4, 'Fourth Factory', '44444 Jenzen Blvd.', 17055, 'Pittsburgh', 'PA' ); COMMIT ; 执行下面的语句向源数据库的daily_production表添加数据。sql view plain copy 在CODE上查看代码片派生到我的代码片USE source; set ye

23、sterday:=date_sub(current_date, interval 1 day); INSERT INTO daily_production VALUES (1, yesterday, 4, 100 ) , (2, yesterday, 3, 200 ) , (3, yesterday, 2, 300 ) , (4, yesterday, 1, 400 ) , (1, yesterday, 1, 400 ) , (2, yesterday, 2, 300 ) , (3, yesterday, 3, 200 ) , (4, yesterday, 4, 100 ); COMMIT; 现在已经做好了测试产品定期装载的准备，使用下面的命令执行定期装载作业。plain view plain copy 在CODE上查看代码片派生到我的代码片./regular_etl_daily_production.sh 使用下面的SQL语句查询production_fact表，确认每天产品数据的定期装载是正确的。sql view plain copy 在CODE上查看代码片派生到我的代码片select product_sk product_sk, production_date_sk date_sk, factory_sk fac