--设置scd的生效时间和过期时间

SET hivevar:cur_date = CURRENT_DATE();SET hivevar:pre_date = DATE_ADD(${hivevar:cur_date},-1);SET hivevar:max_date = CAST('2050-01-01' ASDATE);--设置cdc的开始结束日期

INSERT overwrite TABLErds.cdc_timeSELECT last_load, ${hivevar:cur_date} FROMrds.cdc_time;--装载customer维度--获取源数据中被删除的客户和地址发生改变的客户，将这些数据设置为过期时间，即当前时间的前一天

UPDATEdim_customerSET expiry_date =${hivevar:pre_date}WHERE dim_customer.customer_sk IN(SELECTa.customer_skFROM (SELECTcustomer_sk,

customer_number,

customer_street_addressFROMdim_customerWHERE expiry_date =${hivevar:max_date}) aLEFT JOIN rds.customer b ON a.customer_number =b.customer_numberWHERE b.customer_number IS NULL

OR a.customer_street_address <>b.customer_street_address);--将有地址变化的插入到dim_customer表，如果有相同数据存在有不过期的数据则不插入

INSERT INTOdim_customerSELECT 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_dateFROM(SELECTt2.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`,

${hivevar:pre_date} effective_date,

${hivevar:max_date} expiry_dateFROMdim_customer t1INNER JOIN rds.customer t2 ON t1.customer_number =t2.customer_numberAND t1.expiry_date =${hivevar:pre_date}LEFT JOIN dim_customer t3 ON t1.customer_number =t3.customer_numberAND t3.expiry_date =${hivevar:max_date}WHERE t1.customer_street_address <>t2.customer_street_addressAND t3.customer_sk IS NULL) t1CROSS JOIN(SELECT

COALESCE(MAX(customer_sk),0) sk_maxFROMdim_customer) t2;--处理customer_name列上的scd1，覆盖--不进行更新，将源数据中的name列有变化的数据提取出来，放入临时表--将 dim_couster中这些数据删除、--将临时表中的数据插入

DROP TABLE IF EXISTStmp;CREATE TABLE tmp AS

SELECTa.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_dateFROMdim_customer aJOIN rds.customer b ON a.customer_number =b.customer_numberAND(a.customer_name <>b.customer_name);--删除数据

DELETE FROMdim_customerWHEREdim_customer.customer_skIN (SELECT customer_sk FROMtmp);--插入数据

INSERT INTOdim_customerSELECT * FROMtmp;--处理新增的customer记录

INSERT INTOdim_customerSELECT 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,

${hivevar:pre_date},

${hivevar:max_date}FROM( SELECT t1.*

FROMrds.customer t1LEFT JOIN dim_customer t2 ON t1.customer_number =t2.customer_numberWHERE t2.customer_sk IS NULL) t1CROSS JOIN(SELECT

COALESCE(MAX(customer_sk),0) sk_maxFROMdim_customer) t2;--装载product维度--取源数据中删除或者属性发生变化的产品，将对应

UPDATEdim_productSET expiry_date =${hivevar:pre_date}WHERE dim_product.product_sk IN(SELECTa.product_skFROM(SELECTproduct_sk,

product_code,

product_name,

product_categoryFROMdim_productWHERE expiry_date =${hivevar:max_date}) aLEFT JOIN rds.product b ON a.product_code =b.product_codeWHERE b.product_code IS NULL

OR (a.product_name <> b.product_name OR a.product_category <>b.product_category));--处理product_name、product_category列上scd2的新增行

INSERT INTOdim_productSELECT row_number() over (ORDER BY t1.product_code) +t2.sk_max,

t1.product_code,

t1.product_name,

t1.product_category,

t1.version,

t1.effective_date,

t1.expiry_dateFROM( SELECTt2.product_code product_code,

t2.product_name product_name,

t2.product_category product_category,

t1.version+ 1`version`,

${hivevar:pre_date} effective_date,

${hivevar:max_date} expiry_dateFROMdim_product t1INNER JOIN rds.product t2 ON t1.product_code =t2.product_codeAND t1.expiry_date =${hivevar:pre_date}LEFT JOIN dim_product t3 ON t1.product_code =t3.product_codeAND t3.expiry_date =${hivevar:max_date}WHERE(t1.product_name <>t2.product_nameOR t1.product_category <>t2.product_category)AND t3.product_sk IS NULL) t1CROSS JOIN (SELECT COALESCE(MAX(product_sk),0) sk_maxFROMdim_product) t2;--处理新增的 product 记录

INSERT INTOdim_productSELECT row_number() over (ORDER BY t1.product_code) +t2.sk_max,

t1.product_code,

t1.product_name,

t1.product_category,1,

${hivevar:pre_date},

${hivevar:max_date}FROM( SELECT t1.*

FROMrds.product t1LEFT JOIN dim_product t2 ON t1.product_code =t2.product_codeWHERE t2.product_sk IS NULL) t1CROSS JOIN (SELECT COALESCE(MAX(product_sk),0) sk_maxFROMdim_product) t2;--装载order维度

INSERT INTOdim_orderSELECT row_number() over (ORDER BY t1.order_number) +t2.sk_max,

t1.order_number,

t1.version,

t1.effective_date,

t1.expiry_dateFROM( SELECTorder_number order_number,1`version`,

order_date effective_date,'2050-01-01'expiry_dateFROMrds.sales_order, rds.cdc_timeWHERE entry_date >= last_load AND entry_date

INSERT INTOsales_fact_sales_orderSELECTorder_sk,

customer_sk,

product_sk,

date_sk,

order_amountFROMrds.sales_order a,

dim_order b,

dim_customer c,

dim_product d,

date_dim e,

rds.cdc_time fWHERE a.order_number =b.order_numberAND a.customer_number =c.customer_numberAND a.order_date >=c.effective_dateAND a.order_date =d.effective_dateAND a.order_date =f.last_loadAND a.entry_date

INSERT overwrite TABLErds.cdc_timeSELECTcurrent_load, current_loadFROM rds.cdc_time;