Dataworks数据建模

cd /root/init_table_task/
/usr/bin/python3.6 table_2_odps.py AKID AKSecret 项目名称

https://dataworks.console.aliyun.com/overview

set odps.sql.type.system.odps2=true;
set odps.sql.decimal.odps2=true;
-- 租赁事实表
CREATE TABLE IF NOT EXISTS dwd_fct_rental
(
rental_id BIGINT
,staff_key BIGINT
,customer_key BIGINT
,film_key BIGINT
,store_key BIGINT
,rental_date_key STRING
,rental_counts BIGINT
,inventory_id BIGINT
,etl_date DATETIME
) PARTITIONED BY (ds STRING);
-- 归还事实表
CREATE TABLE IF NOT EXISTS dwd_fct_return
(
rental_id BIGINT
,customer_key BIGINT
,film_key BIGINT
,store_key BIGINT
,return_date_key STRING
,return_counts BIGINT
,inventory_id BIGINT
,etl_date DATETIME
) PARTITIONED BY (ds STRING);
-- 累积快照事实表
CREATE TABLE IF NOT EXISTS dwd_fct_order
(
rental_id BIGINT
,staff_key BIGINT
,customer_key BIGINT
,film_key BIGINT
,store_key BIGINT
,rental_date_key STRING
,return_date_key STRING
,rental_counts BIGINT
,return_counts BIGINT
,inventory_id BIGINT
,etl_date DATETIME
) PARTITIONED BY (ds STRING);
-- 电影维度表
CREATE TABLE IF NOT EXISTS dim_film
(
film_key BIGINT
,film_title STRING
,film_description STRING
,film_release_year BIGINT
,film_language STRING
,film_rental_duration BIGINT
,film_rental_rate DECIMAL(4,2)
,film_duration BIGINT
,film_replacement_cost DECIMAL(6,2)
,film_rating_text STRING
,film_category STRING
,film_last_update DATETIME
,etl_date DATETIME
) PARTITIONED BY (ds STRING);
-- 日期维度表
CREATE TABLE IF NOT EXISTS dim_date
(
date_key STRING
,date_value DATE
,year_value STRING
,month_value STRING
,week_value STRING
,day_value STRING
,etl_date DATETIME
);
-- 客户维度表
CREATE TABLE IF NOT EXISTS dim_customer
(
customer_key BIGINT
,customer_first_name STRING
,customer_last_name STRING
,customer_active STRING
,customer_address STRING
,customer_district STRING
,customer_city STRING
,customer_country STRING
,customer_last_update DATETIME
,etl_date DATETIME
) PARTITIONED BY (ds STRING);
-- 店铺维度表
CREATE TABLE IF NOT EXISTS dim_store
(
store_key BIGINT
,store_address STRING
,store_district STRING
,store_city STRING
,store_country STRING
,store_manager_staff_id BIGINT
,store_manager_first_name STRING
,store_manager_last_name STRING
,store_last_update DATETIME
,etl_date DATETIME
) PARTITIONED BY (ds STRING);
-- 店员维度表
CREATE TABLE IF NOT EXISTS dim_staff
(
staff_key BIGINT
,staff_store_id BIGINT
,staff_first_name STRING
,staff_last_name STRING
,staff_active STRING
,staff_last_update DATETIME
,etl_date DATETIME
) PARTITIONED BY (ds STRING);

INSERT OVERWRITE TABLE dwd_fct_rental PARTITION(ds = '${bizdate}')
SELECT
r.rental_id as rental_id
,r.staff_id as staff_key
,r.customer_id as customer_key
,i.film_id as film_key
,i.store_id as store_key
,to_char(r.rental_date, 'yyyymmdd') as rental_date_key
,1 as rental_counts
,r.inventory_id as inventory_id
,GETDATE() as etl_date
FROM ods_sakila_rental_df r
LEFT JOIN ods_sakila_inventory_df i
ON r.inventory_id = i.inventory_id
AND i.ds = '${bizdate}'
WHERE r.ds = '${bizdate}'
;

INSERT OVERWRITE TABLE dwd_fct_return PARTITION(ds = '${bizdate}')
SELECT
r.rental_id as rental_id
,r.customer_id as customer_key
,i.film_id as film_key
,i.store_id as store_key
,to_char(r.return_date, 'yyyymmdd') as return_date_key
,1 as return_counts
,r.inventory_id as inventory_id
,GETDATE() as etl_date
FROM ods_sakila_rental_df r
INNER JOIN ods_sakila_inventory_df i
ON r.inventory_id = i.inventory_id
AND i.ds = '${bizdate}'
AND r.return_date IS NOT NULL
WHERE r.ds = '${bizdate}'
;

INSERT OVERWRITE TABLE dwd_fct_order PARTITION(ds = '${bizdate}')
SELECT
r.rental_id as rental_id
,r.staff_id as staff_key
,r.customer_id as customer_key
,i.film_id as film_key
,i.store_id as store_key
,to_char(r.rental_date, 'yyyymmdd') as rental_date_key
,to_char(r.return_date, 'yyyymmdd') as return_date_key
,1 as rental_counts
,case when i.inventory_id is not null
then 1 else 0
end as return_counts
,r.inventory_id as inventory_id
,GETDATE() as etl_date
FROM ods_sakila_rental_df r
LEFT JOIN ods_sakila_inventory_df i
ON r.inventory_id = i.inventory_id
AND i.ds = '${bizdate}'
WHERE r.ds = '${bizdate}'
;

set odps.sql.decimal.odps2=true;
INSERT OVERWRITE TABLE dim_film PARTITION(ds = '${bizdate}')
SELECT
f.film_id as film_key
,f.title as film_title
,f.description as film_description
,cast(f.release_year AS BIGINT ) as film_release_year
,l.name as film_language
,f.rental_duration as film_rental_duration
,cast(f.rental_rate AS DECIMAL(4,2)) as film_rental_rate
,f.length as film_duration
,cast(f.replacement_cost AS DECIMAL(6,2)) as film_replacement_cost
,f.rating as film_rating_text
,c.name as film_category
,f.last_update as film_last_update
,GETDATE() as etl_date
FROM ods_sakila_film_df f
LEFT JOIN ods_sakila_language_df l
ON f.language_id = l.language_id
AND l.ds = '${bizdate}'
LEFT JOIN ods_sakila_film_category_df fc
ON f.film_id = fc.film_id
AND fc.ds = '${bizdate}'
LEFT JOIN ods_sakila_category_df c
ON fc.category_id = c.category_id
AND c.ds = '${bizdate}'
WHERE f.ds = '${bizdate}'
;

set odps.sql.type.system.odps2=true;
INSERT OVERWRITE TABLE dim_date
SELECT
TO_CHAR(dt, 'yyyymmdd') as date_key
,dt as date_value
,SUBSTRING(dt,1,4) as year_value
,SUBSTRING(dt,6,2) as month_value
,WEEKOFYEAR(dt) as week_value
,SUBSTRING(dt,9,2) as day_value
,GETDATE() as etl_date
FROM (
SELECT DATE_ADD('2005-01-01', cast(rn AS BIGINT)) AS dt
FROM (
SELECT transform(
'for i in `seq 0 7000`; do echo $i; done'
) USING 'sh' AS rn
) a
) b
;

INSERT OVERWRITE TABLE dim_customer PARTITION(ds = '${bizdate}')
SELECT
c.customer_id as customer_key
,c.first_name as customer_first_name
,c.last_name as customer_last_name
,c.active as customer_active
,a.address as customer_address
,a.district as customer_district
,ct.city as customer_city
,cy.country as customer_country
,c.last_update as customer_last_update
,GETDATE() as etl_date
FROM ods_sakila_customer_df c
LEFT JOIN ods_sakila_address_df a
ON c.address_id = a.address_id
AND a.ds = '${bizdate}'
LEFT JOIN ods_sakila_city_df ct
ON a.city_id = ct.city_id
AND ct.ds = '${bizdate}'
LEFT JOIN ods_sakila_country_df cy
ON ct.country_id = cy.country_id
AND cy.ds = '${bizdate}'
WHERE c.ds = '${bizdate}'
;

INSERT OVERWRITE TABLE dim_store PARTITION(ds = '${bizdate}')
SELECT
s.store_id as store_key
,a.address as store_address
,a.district as store_district
,ct.city as store_city
,cy.country as store_country
,s.manager_staff_id as store_manager_staff_id
,st.first_name as store_manager_first_name
,st.last_name as store_manager_last_name
,s.last_update as store_last_update
,GETDATE() as etl_date
FROM ods_sakila_store_df s
LEFT JOIN ods_sakila_address_df a
ON s.address_id = a.address_id
AND a.ds = '${bizdate}'
LEFT JOIN ods_sakila_city_df ct
ON a.city_id = ct.city_id
AND ct.ds = '${bizdate}'
LEFT JOIN ods_sakila_country_df cy
ON ct.country_id = cy.country_id
AND cy.ds = '${bizdate}'
LEFT JOIN ods_sakila_staff_df st
ON s.manager_staff_id = st.staff_id
WHERE s.ds = '${bizdate}'
AND st.ds = '${bizdate}'
;

INSERT OVERWRITE TABLE dim_staff PARTITION(ds = '${bizdate}')
SELECT s.staff_id as staff_key
,s.store_id as staff_store_id
,s.first_name as staff_first_name
,s.last_name as staff_last_name
,s.active as staff_active
,s.last_update as staff_last_update
,GETDATE() as etl_date
FROM ods_sakila_staff_df s
WHERE ds = '${bizdate}'
;