Rebuild entire (small) DWH in one go

examples/hive-example/dags/process_hive_dwh.py

@@ -16,15 +16,15 @@
 import airflow
 from datetime import datetime, timedelta
 from airflow.operators.hive_operator import HiveOperator
-from airflow.operators.dummy_operator import DummyOperator
+from airflow.operators.latest_only_operator import LatestOnlyOperator
 from airflow.models import Variable
 
 
 args = {
     'owner': 'airflow',
     'start_date': airflow.utils.dates.days_ago(7),
     'provide_context': True,
-    'depends_on_past': True
+    'depends_on_past': False
 }
 
 tmpl_search_path = Variable.get("hive_sql_path")
@@ -47,7 +47,7 @@
     dagrun_timeout=timedelta(minutes=60),
     template_searchpath=tmpl_search_path,
     default_args=args,
-    max_active_runs=1)
+    max_active_runs=10)
 
 customer_step_1 = HiveOperator(
     hql='customer/step_1.hql',
@@ -65,62 +65,6 @@
     task_id='customer_step_2',
     dag=dag)
 
-customer_step_3 = HiveOperator(
-    hql='customer/step_3.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='customer_step_3',
-    dag=dag)
-
-customer_step_4 = HiveOperator(
-    hql='customer/step_4.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='customer_step_4',
-    dag=dag)
-
-customer_step_5 = HiveOperator(
-    hql='customer/step_5.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='customer_step_5',
-    dag=dag)
-
-customer_step_6 = HiveOperator(
-    hql='customer/step_6.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='customer_step_6',
-    dag=dag)
-
-customer_step_7 = HiveOperator(
-    hql='customer/step_7.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='customer_step_7',
-    dag=dag)
-
-customer_step_8 = HiveOperator(
-    hql='customer/step_8.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='customer_step_8',
-    dag=dag)
-
-customer_step_9 = HiveOperator(
-    hql='customer/step_9.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='customer_step_9',
-    dag=dag)
-
 
 product_step_1 = HiveOperator(
     hql='product/step_1.hql',
@@ -138,76 +82,36 @@
     task_id='product_step_2',
     dag=dag)
 
-product_step_3 = HiveOperator(
-    hql='product/step_3.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='product_step_3',
+latest_only = LatestOnlyOperator(
+    task_id='latest_only',
     dag=dag)
 
-product_step_4 = HiveOperator(
-    hql='product/step_4.hql',
+fact_order_step_1 = HiveOperator(
+    hql='order/step_1.hql',
     hive_cli_conn_id='hive_staging',
     schema='default',
     hiveconf_jinja_translate=True,
-    task_id='product_step_4',
+    task_id='fact_order_step_1',
     dag=dag)
 
-product_step_5 = HiveOperator(
-    hql='product/step_5.hql',
+fact_order_step_2 = HiveOperator(
+    hql='order/step_2.hql',
     hive_cli_conn_id='hive_staging',
     schema='default',
     hiveconf_jinja_translate=True,
-    task_id='product_step_5',
+    task_id='fact_order_step_2',
     dag=dag)
 
-product_step_6 = HiveOperator(
-    hql='product/step_6.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='product_step_6',
-    dag=dag)
-
-product_step_7 = HiveOperator(
-    hql='product/step_7.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='product_step_7',
-    dag=dag)
-
-product_step_8 = HiveOperator(
-    hql='product/step_8.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='product_step_8',
-    dag=dag)
-
-product_step_9 = HiveOperator(
-    hql='product/step_9.hql',
-    hive_cli_conn_id='hive_staging',
-    schema='default',
-    hiveconf_jinja_translate=True,
-    task_id='product_step_9',
-    dag=dag)
-
-dummy = DummyOperator(
-    task_id='dummy',
-    dag=dag)
+latest_only >> customer_step_1
+latest_only >> product_step_1
 
-customer_step_1 >> customer_step_2 >> customer_step_3 >> customer_step_4
-customer_step_4 >> customer_step_5 >> customer_step_6 >> customer_step_7 
-customer_step_7 >> customer_step_8 >> customer_step_9
+customer_step_1 >> customer_step_2
+product_step_1 >> product_step_2
 
-product_step_1 >> product_step_2 >> product_step_3 >> product_step_4
-product_step_4 >> product_step_5 >> product_step_6 >> product_step_7 
-product_step_7 >> product_step_8 >> product_step_9
+customer_step_2 >> fact_order_step_1
+product_step_2 >> fact_order_step_1
 
-customer_step_9 >> dummy
-product_step_9 >> dummy
+fact_order_step_1 >> fact_order_step_2
 
 
 if __name__ == "__main__":

examples/hive-example/dags/staging-oltp.py

@@ -41,7 +41,7 @@
     hive_table='customer_staging',
     postgres_conn_id='postgres_oltp',
     hive_cli_conn_id='hive_staging',
-    partition={"change_date": "{{ ds_nodash }}"},
+    partition={"change_date": "{{ ds }}"},
     parameters={"window_start_date": "{{ ds }}", "window_end_date": "{{ tomorrow_ds }}"},
     task_id='stage_customer',
     dag=dag)
@@ -51,7 +51,7 @@
     hive_table='order_info_staging',
     postgres_conn_id='postgres_oltp',
     hive_cli_conn_id='hive_staging',
-    partition={"change_date": "{{ ds_nodash }}"},
+    partition={"change_date": "{{ ds }}"},
     parameters={"window_start_date": "{{ ds }}", "window_end_date": "{{ tomorrow_ds }}"},
     task_id='stage_orderinfo',
     dag=dag)
@@ -61,7 +61,7 @@
     hive_table='orderline_staging',
     postgres_conn_id='postgres_oltp',
     hive_cli_conn_id='hive_staging',
-    partition={"change_date": "{{ ds_nodash }}"},
+    partition={"change_date": "{{ ds }}"},
     parameters={"window_start_date": "{{ ds }}", "window_end_date": "{{ tomorrow_ds }}"},
     task_id='stage_orderline',
     dag=dag)
@@ -71,7 +71,7 @@
     hive_table='product_staging',
     postgres_conn_id='postgres_oltp',
     hive_cli_conn_id='hive_staging',
-    partition={"change_date": "{{ ds_nodash }}"},
+    partition={"change_date": "{{ ds }}"},
     parameters={"window_start_date": "{{ ds }}", "window_end_date": "{{ tomorrow_ds }}"},
     task_id='stage_product',
     dag=dag)

examples/hive-example/hql/customer/step_1.hql

@@ -1,5 +1,7 @@
 -- Make sure a dimension table exists of the right type
 
+DROP TABLE IF EXISTS dim_customer;
+
 CREATE TABLE IF NOT EXISTS dim_customer (
       dim_customer_key BIGINT
     , customer_id STRING
@@ -9,8 +11,8 @@ CREATE TABLE IF NOT EXISTS dim_customer (
     , scd_version INT -- historical version of the record (1 is the oldest)
     , scd_start_date DATE -- start date
     , scd_end_date DATE -- end date and time (9999-12-31 by default)
-    , scd_active BOOLEAN -- whether it's the latest version or not
 )
+PARTITIONED BY  (scd_active STRING)
 ROW FORMAT DELIMITED
 FIELDS TERMINATED BY '\001'
 STORED AS ORC;

examples/hive-example/hql/customer/step_2.hql

@@ -1,7 +1,21 @@
--- Start by creating a new dimension table, which we use to copy over the production one.
+-- Regenerate entire dimension from scratch.
 
-DROP TABLE IF EXISTS dim_customer_new;
+FROM (SELECT
+          row_number() OVER () AS dim_customer_key
+        , customer_id
+        , LEAD(customer_id) OVER (PARTITION BY customer_id ORDER BY change_date) as id_lead
+        , LAST_VALUE(cust_name) OVER (PARTITION BY customer_id ORDER BY change_date RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) as cust_name
+        , row_number() OVER (PARTITION BY customer_id ORDER BY change_date) AS scd_version
+        , street
+        , city
+        , change_date
+        , LEAD(change_date) OVER (PARTITION BY customer_id ORDER BY change_date) as end_date
+    FROM
+        customer_staging) a
+INSERT OVERWRITE TABLE dim_customer PARTITION(scd_active='T')
+SELECT a.dim_customer_key, a.customer_id, a.cust_name, a.street, a.city, scd_version, TO_DATE(a.change_date) as scd_start_date, TO_DATE('9999-12-31') as scd_end_date
+WHERE a.id_lead IS NULL
 
-CREATE TABLE dim_customer_new
-LIKE dim_customer
-STORED AS ORC;
+INSERT OVERWRITE TABLE dim_customer PARTITION(scd_active='F')
+SELECT a.dim_customer_key, a.customer_id, a.cust_name, a.street, a.city, scd_version, TO_DATE(a.change_date) as scd_start_date, TO_DATE(a.end_date) as scd_end_date
+WHERE a.id_lead = a.customer_id;