基于Hadoop生態(tài)圈的數(shù)據(jù)倉庫實踐 —— 進階技術(shù)(十一)

1、基于Hadoop生態(tài)圈的數(shù)據(jù)倉庫實踐 進階技術(shù)（十一）十一、多重星型模式 從“進階技術(shù)”開始，已經(jīng)通過增加列和表擴展了數(shù)據(jù)倉庫，在進階技術(shù)（五） “快照”里增加了第二個事實表，month_end_sales_order_fact表。這之后數(shù)據(jù)倉庫模式就有了兩個事實表（第一個是在開始建立數(shù)據(jù)倉庫時創(chuàng)建的sales_order_fact表）。有了這兩個事實表的數(shù)據(jù)倉庫就是一個標準的雙星型模式。本節(jié)將在現(xiàn)有的維度數(shù)據(jù)倉庫上再增加一個新的星型結(jié)構(gòu)。與現(xiàn)有的與銷售關(guān)聯(lián)的星型結(jié)構(gòu)不同，新的星型結(jié)構(gòu)關(guān)注的是產(chǎn)品業(yè)務(wù)領(lǐng)域。新的星型結(jié)構(gòu)有一個事實表和一個維度表，用于存儲數(shù)據(jù)倉庫中的產(chǎn)品數(shù)據(jù)。1. 一個新的星型模

2、式 下圖顯示了擴展后的數(shù)據(jù)倉庫模式。模式中有三個星型結(jié)構(gòu)。sales_order_fact表是第一個星型結(jié)構(gòu)的事實表，與其相關(guān)的維度表是customer_dim、product_dim、date_dim和sales_order_attribute_dim表。month_end_sales_order_fact表是第二個星型結(jié)構(gòu)的事實表。product_dim和month_dim是其對應的維度表。第一個和第二個星型結(jié)構(gòu)共享product_dim維度表。第二個星型結(jié)構(gòu)的事實表和月份維度數(shù)據(jù)分別來自于第一個星型結(jié)構(gòu)的事實表和date_dim維度表。它們不從源數(shù)據(jù)獲得數(shù)據(jù)。第三個星型模式的事實表是新建

3、的production_fact表。它的維度除了存儲在已有的date_dim和product_dim表，還有一個新的factory_dim表。第三個星型結(jié)構(gòu)的數(shù)據(jù)來自源數(shù)據(jù)。 執(zhí)行下面的腳本建立第三個星型模式中的新表和對應的源數(shù)據(jù)表。sql view plain copy 在CODE上查看代碼片派生到我的代碼片- 在MySQL源庫上建立工廠表和每日產(chǎn)品表 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）新建抽取作業(yè)plain view plain copy 在CODE上查看

9、代碼片派生到我的代碼片# 建立增量抽取每日產(chǎn)品表的作業(yè)，以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 新建定期裝載每日產(chǎn)品腳本文件regular_etl_daily_production.sh，內(nèi)容如下。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 # 增量抽取每日產(chǎn)品表 sqoop job -exec myjob_incremental_import_daily_production -meta-connect jdbc:hsqldb:hsql:/cdh2:16000/sqoop # 調(diào)用regular_etl_daily_production.sql文件執(zhí)行定期裝載 beeline -u jdbc:hive2:/cdh2:10000/dw -f regular_etl_daily_production.sql 為了和其

13、它定期裝載腳本共用環(huán)境和時間窗口設(shè)置，新建一個set_time.sql腳本，內(nèi)容如下。sql view plain copy 在CODE上查看代碼片派生到我的代碼片- 設(shè)置變量以支持事務(wù) 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; - 設(shè)置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); - 設(shè)置CDC的上限時間 INSERT OVERWRITE TABLE rds.cdc_time SELECT last_load, $hivevar:cur_date FROM rds.cdc_time; 新建regular_et

15、l_daily_production.sql腳本文件內(nèi)容如下。sql view plain copy 在CODE上查看代碼片派生到我的代碼片- 設(shè)置環(huán)境與時間窗口 !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; - 裝載每日產(chǎn)品事實表 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. 測試 到目前為止已經(jīng)討論了第三個星型結(jié)構(gòu)里的所有表，現(xiàn)在做一些測試。首先需要一些工廠信息。執(zhí)行下面的語句向源數(shù)據(jù)庫的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 ; 執(zhí)行下面的語句向源數(shù)據(jù)庫的daily_production表添加數(shù)據(jù)。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; 現(xiàn)在已經(jīng)做好了測試產(chǎn)品定期裝載的準備，使用下面的命令執(zhí)行定期裝載作業(yè)。plain view plain copy 在CODE上查看代碼片派生到我的代碼片./regular_etl_daily_production.sh 使用下面的SQL語句查詢production_fact表，確認每天產(chǎn)品數(shù)據(jù)的定期裝載是正確的。sql view plain copy 在CODE上查看代碼片派生到我的代碼片select product_sk product_sk, production_date_sk date_sk, factory_sk fac