changes to create demo for dbt docs (#32)

Author: entechlog

dbt-docker/docker/dbt/requirements.txt

@@ -1,6 +1,6 @@
-dbt-core==1.0.1
-dbt-postgres==1.0.1
-dbt-redshift==1.0.0
-dbt-snowflake==1.0.0
-dbt-bigquery==1.0.0
-dbt-materialize==1.0.1
+dbt-core
+dbt-postgres
+dbt-redshift
+dbt-snowflake
+dbt-bigquery
+dbt-materialize

dbt-lambda/.env.template

@@ -0,0 +1,4 @@
+export ENV_CODE="DEV"
+export PROJ_CODE="ENTECHLOG"
+export SNOWSQL_ACCOUNT=""
+export SNOWSQL_PWD=""

dbt-lambda/README.md

@@ -0,0 +1,54 @@
+# Lambda Views Demo
+
+This repository demonstrates a **Lambda Architecture** using dbt. It shows how you can combine historical and near real-time data through separate layers.
+
+![dbt Docs](./assets/dbt-docs.png)
+
+## What’s Included
+
+- **Prep Layer:**  
+  Loads and prepares all raw data from the `purchases` seed. This layer supports querying all data from the raw seed.
+
+- **Fact Layer Macro:**  
+  A macro (`fact_purchases_sql`) that transforms the prep data and applies a date filter using the `lambda_filter_by_date` macro.
+
+- **Fact Layer Model:**  
+  A model that calls the fact layer macro to build an incremental fact table from the prep data.
+
+- **OBT/Lambda View:**  
+  A view that uses `lambda_union_by_date` to dynamically union historical fact data with new raw data based on the batch cycle date. It determines whether to pull data from the fact or raw layer.
+
+## Steps to Run the Demo
+
+1. **Seed the Data:**  
+   Run the following command to load the raw seed data:
+   ```bash
+   dbt seed
+   ```
+
+2. **Build the Models:**  
+   Build all models by running:
+   ```bash
+   dbt run
+   ```
+
+3. **Initial Check:**  
+   - Query the **fact** model and the **OBT/lambda view**.
+   - You should see **20 records** in each.
+
+4. **Simulate New Data:**  
+   - Move the data from the backup seed into the raw seed file (this simulates new data arriving).
+   - Run the seed command again:
+     ```bash
+     dbt seed
+     ```
+
+5. **Rebuild and Validate:**  
+   Run:
+   ```bash
+   dbt run
+   ```
+   - Now, when you query the **fact** model, it should still show **20 records**.
+   - The **OBT/lambda view** will display **45 records**, with **25 new records** coming from the raw layer.
+
+This demo shows how the Lambda Architecture in dbt can dynamically merge historical data with new incoming data at query time.

dbt-lambda/assets/dbt-docs.png

@@ -0,0 +1,43 @@
+name: 'dbt_lambda_demo'
+version: '1.0.0'
+config-version: 2
+
+profile: 'dbt-snowflake'
+
+vars:
+  run_type: "daily"
+  batch_cycle_date: '2024-01-01'
+
+source-paths: ["models"]
+macro-paths: ["macros"]
+seed-paths: ["seeds"]
+
+models:
+  dbt_lambda_demo:
+    +materialized: table
+    prep:
+      dim:
+        database: "{{ env_var('ENV_CODE') | trim }}_{{ env_var('PROJ_CODE') | trim }}_PREP_DB"
+        schema: DIM
+        materialized: view
+      fact:
+        database: "{{ env_var('ENV_CODE') | trim }}_{{ env_var('PROJ_CODE') | trim }}_PREP_DB"
+        schema: FACT
+        materialized: view
+    dw:
+      dim:
+        database: "{{ env_var('ENV_CODE') | trim }}_{{ env_var('PROJ_CODE') | trim }}_DW_DB"
+        schema: DIM
+        materialized: view
+      fact:
+        database: "{{ env_var('ENV_CODE') | trim }}_{{ env_var('PROJ_CODE') | trim }}_DW_DB"
+        schema: FACT
+        materialized: view
+    obt:
+      database: "{{ env_var('ENV_CODE') | trim }}_{{ env_var('PROJ_CODE') | trim }}_DW_DB"
+      schema: OBT
+
+seeds:
+  dbt_lambda_demo:
+    database: "{{ env_var('ENV_CODE') | trim }}_{{ env_var('PROJ_CODE') | trim }}_PREP_DB"
+    schema: SEED

dbt-lambda/dbt_project.yml

@@ -0,0 +1,20 @@
+{% macro lambda_filter_by_date(src_column_name, tgt_column_name) %}
+
+    {% set materialized = config.require('materialized') %}
+    {% set filter_time = var('lambda_split', run_started_at) %}
+
+    {% if materialized == 'view' %}
+
+        where {{ src_column_name }} >= (select max({{ tgt_column_name }}) from {{ this | replace('.OBT.', '.FACT.') }})
+
+    {% elif is_incremental() %}
+
+        where DATE({{ src_column_name }}) = '{{ var("batch_cycle_date") }}'
+
+    {% else %}
+
+        where {{ src_column_name }} < '{{ filter_time }}'
+
+    {% endif %}
+
+{% endmacro %}

dbt-lambda/macros/lambda/lambda_filter_by_date.sql

@@ -0,0 +1,51 @@
+{% macro lambda_union_by_date(historical_relation, model_sql,tgt_timestamp_column) %}
+
+{% set unique_key = config.get('unique_key', none) %}
+
+with historical as (
+
+    select *
+      {#  ,'dw' as _dbt_lambda_view_source, #}
+       {#  ({{ get_max_timestamp(historical_relation,tgt_timestamp_column) }}) as _dbt_last_run_at #}
+
+    from {{ historical_relation }}
+
+),
+
+new_raw as (
+
+    {{ model_sql }}
+
+),
+
+new as (
+
+    select *
+       {# ,'raw' as _dbt_lambda_view_source, #}
+      {# {{ dbt_utils.current_timestamp() }} as _dbt_last_run_at #}
+
+    from new_raw
+
+),
+
+unioned as (
+
+    select * from historical
+
+    {% if unique_key %}
+
+        where {{ unique_key }} not in (
+            select {{ unique_key }} from new
+        )
+
+    {% endif %}
+
+    union
+
+    select * from new
+
+)
+
+select * from unioned
+
+{% endmacro %}

dbt-lambda/macros/lambda/lambda_union_by_date.sql

@@ -0,0 +1,28 @@
+{% macro fact_purchases_sql() %}
+
+WITH source AS (
+    SELECT *
+    FROM {{ ref('prep__fact_purchases') }}
+         {{ lambda_filter_by_date(src_column_name='event_timestamp', tgt_column_name='event_timestamp') }}
+),
+
+relations AS (
+    SELECT 
+        event_date,
+        event_hour,
+        event_timestamp,
+        store_key,
+        product_key,
+        product_name,
+        quantity,
+        unit_price,
+        extended_price,
+        payment_method,
+        purchase_key
+    FROM source
+)
+
+SELECT *
+FROM relations
+
+{% endmacro %}

dbt-lambda/macros/models/fact_purchases_sql.sql

@@ -0,0 +1,41 @@
+{% macro delete_data(del_key, del_value, this, offset_days=0) %}
+
+{%- set relation = adapter.get_relation(this.database, this.schema, this.table) -%}
+
+{%- if relation is not none -%}
+	{%- if var("run_type") == 'full-refresh'-%}
+			{%- call statement('truncate table ' ~ this, fetch_result=False, auto_begin=True) -%}
+				truncate {{ relation }}
+			{%- endcall -%}
+	{%- elif var("run_type") == 'daily'-%}	
+		{% if offset_days != 0 %}
+			{%- set sub_days = offset_days | default(0) -%}
+			{%- set del_value_datetime = modules.datetime.datetime.strptime(del_value, "%Y%m%d") + modules.datetime.timedelta(days=sub_days) -%}
+			{%- set del_value_date = del_value_datetime.strftime("'%Y%m%d'") -%}
+			{%- call statement('delete ' ~ del_value_date ~ ' records from ' ~ this, fetch_result=False, auto_begin=True) -%}
+				delete from {{ relation }} where {{del_key}} = {{del_value_date}}
+			{%- endcall -%}
+		{% else %}
+			{%- call statement('delete ' ~ del_value ~ ' records from ' ~ this, fetch_result=False, auto_begin=True) -%}
+				delete from {{ relation }} where {{del_key}} = '{{del_value}}'
+			{%- endcall -%}
+		{% endif %}
+	{%- elif var("run_type") == 'backfill'-%}
+		{% if offset_days > 0 %}
+			{%- set sub_days = offset_days | default(0) -%}
+			{%- set backfill_start_datetime = modules.datetime.datetime.strptime( var("backfill_start_date") , "%Y-%m-%d") + modules.datetime.timedelta(days=sub_days) -%}
+			{%- set backfill_end_datetime = modules.datetime.datetime.strptime(var("backfill_end_date") , "%Y-%m-%d") + modules.datetime.timedelta(days=sub_days) -%}
+			{%- set backfill_start_date = backfill_start_datetime.strftime("'%Y%m%d'") -%} 
+			{%- set backfill_end_date = backfill_end_datetime.strftime("'%Y%m%d'") -%} 
+			{%- call statement('delete ' ~ del_value ~ ' records from ' ~ this ~ ' where '~del_key~ ' between ' ~ backfill_start_date ~ ' and ' ~ backfill_end_date, fetch_result=False, auto_begin=True) -%}
+				delete from {{ relation }} where {{del_key}} between {{backfill_start_date}} and {{backfill_end_date}}
+			{%- endcall -%}
+		{% else %}
+			{%- call statement('delete ' ~ del_value ~ ' records from ' ~ this ~ ' where '~del_key~ ' between ' ~ var("backfill_start_date") ~ ' and ' ~ var("backfill_end_date"), fetch_result=False, auto_begin=True) -%}
+				delete from {{ relation }} where {{ del_key }} between '{{ var("backfill_start_date") | replace('-', '') }}' and '{{ var("backfill_end_date") | replace('-', '') }}'
+			{%- endcall -%}
+		{% endif %}
+	{% endif %}
+{%- endif -%}
+
+{% endmacro %}

dbt-lambda/macros/utils/delete_data.sql

@@ -0,0 +1,14 @@
+{% macro generate_database_name(custom_database_name=none, node=none) -%}
+
+    {%- set default_database = target.database -%}
+    {%- if custom_database_name is none -%}
+
+        {{ default_database }}
+
+    {%- else -%}
+
+        {{ custom_database_name | trim }}
+
+    {%- endif -%}
+
+{%- endmacro %}