Temp

WIP revise models and point to duckdb file

Update

Update

Author: MajaCernja

channel_messages_stats.csv

@@ -7,6 +7,7 @@
 """
 
 import argparse
+import json
 import random
 from pathlib import Path
 
@@ -178,13 +179,9 @@ def process_csv(input_path: Path, output_path: Path) -> None:
     """
     Read input CSV, generate reactions for each row, and write output CSV.
     """
-    # Read input CSV
-    df = pd.read_csv(input_path)
-    
-    # Generate reactions for each row
-    df['reactions'] = [str(generate_reactions()) for _ in range(len(df))]
 
-    # Write output CSV
+    df = pd.read_csv(input_path)
+    df['reactions'] = [json.dumps(generate_reactions()) for _ in range(len(df))]
     df.to_csv(output_path, index=False)
 
     print(f"Processed {len(df)} rows")

scripts_python/generate_channel_messages_reactions.py

@@ -1,7 +1,27 @@
 {% macro to_date(timestamp, localize=True, timezone=var('local_timezone')) %}
+    {{ return(adapter.dispatch('to_date', 'slack_analytics')(timestamp, localize, timezone)) }}
+{% endmacro %}
+
+{% macro default__to_date(timestamp, localize, timezone) %}
     {% if localize %}
     TO_DATE(FROM_UTC_TIMESTAMP({{ timestamp }}, '{{ timezone }}'))
     {% else %}
     TO_DATE({{ timestamp }})
     {% endif %}
-{% endmacro %}
+{% endmacro %}
+
+{% macro duckdb__to_date(timestamp, localize, timezone) %}
+    {% if localize %}
+    CAST(timezone('{{ timezone }}', {{ timestamp }}::TIMESTAMPTZ) AS DATE)
+    {% else %}
+    CAST({{ timestamp }} AS DATE)
+    {% endif %}
+{% endmacro %}
+
+{% macro snowflake__to_date(timestamp, localize, timezone) %}
+    {% if localize %}
+    TO_DATE(CONVERT_TIMEZONE('UTC', '{{ timezone }}', {{ timestamp }}))
+    {% else %}
+    TO_DATE({{ timestamp }})
+    {% endif %}
+{% endmacro %}

shared/projects/dbt/slack_analytics/macros/to_date.sql

@@ -11,5 +11,5 @@
 {% endmacro %}
 
 {% macro duckdb__unnest_array(array_column, alias) %}
-    cross join unnest({{ array_column }}) as t({{ alias }})
+    , unnest(from_json({{ array_column }}, '["JSON"]')) as t({{ alias }})
 {% endmacro %}

shared/projects/dbt/slack_analytics/macros/unnest_array.sql

@@ -34,6 +34,6 @@ models:
       - name: message_datetime
         data_type: timestamp
         description: The date and time the message was sent.
-      - name: extracted_at
+      - name: extracted_datetime
         data_type: timestamp
         description: The date and time the message was extracted from the Slack API.

shared/projects/dbt/slack_analytics/models/dimensions/_models.yml

@@ -3,7 +3,7 @@
         materialized='incremental',
         unique_key='message_id',
         partition_by=['message_date'],
-        incremental_strategy='merge'
+        incremental_strategy='delete+insert'
     )
 }}
 
@@ -20,11 +20,11 @@ with staging as (
         reactions,
         {{ to_date('message_datetime', localize=True, timezone=var('local_timezone')) }} as message_date,
         message_datetime,
-        extracted_at
+        extracted_datetime
     from {{ ref('stg_channel_messages') }}
     {% if is_incremental() %}
-        where extracted_at > (
-            select max(extracted_at) from {{ this }}
+        where extracted_datetime > (
+            select max(extracted_datetime) from {{ this }}
         )
     {% endif %}        
 )
@@ -42,11 +42,11 @@ with staging as (
         reactions,
         message_date,
         message_datetime,
-        extracted_at
+        extracted_datetime
 
     from staging 
     where 1=1
-    qualify row_number() over (partition by message_id order by extracted_at desc) = 1
+    qualify row_number() over (partition by message_id order by extracted_datetime desc) = 1
 )
 
 select * from dimension

shared/projects/dbt/slack_analytics/models/dimensions/dim_slack_messages.sql

@@ -31,6 +31,6 @@ models:
       - name: reaction_user
         data_type: string
         description: The user_id of the user who used the specified reaction to the message.
-      - name: extracted_at
+      - name: extracted_datetime
         data_type: timestamp
         description: The date and time the message was extracted from the Slack API.

shared/projects/dbt/slack_analytics/models/facts/_models.yml

@@ -3,16 +3,16 @@
         materialized='incremental',
         unique_key='message_reaction_id',
         partition_by=['message_date'],
-        incremental_strategy='merge'
+        incremental_strategy='delete+insert'
     )
 }}
 
 -- get new data from staging
 with dim_messages as (
     select * from {{ ref('dim_slack_messages') }}
     {% if is_incremental() %}
-        where extracted_at > (
-            select max(extracted_at) from {{ this }}
+        where extracted_datetime > (
+            select max(extracted_datetime) from {{ this }}
         )
     {% endif %}
 )
@@ -25,7 +25,7 @@ with dim_messages as (
         reaction.name as reaction_name,
         message_date,
         message_datetime,
-        extracted_at
+        extracted_datetime
     from dim_messages
     {{ unnest_array('reactions', 'reaction') }}
 )
@@ -38,7 +38,7 @@ with dim_messages as (
         reaction_user,
         message_date,
         message_datetime,
-        extracted_at
+        extracted_datetime
     from semi_expanded_reactions
     {{ unnest_array('reaction_users', 'reaction_user') }}
 )
@@ -57,7 +57,7 @@ with dim_messages as (
             else reaction_name 
             end as reaction_name_normalised,
         reaction_user,
-        extracted_at
+        extracted_datetime  
     from expanded_reactions s
 )
 

shared/projects/dbt/slack_analytics/models/facts/fct_slack_message_reactions.sql

@@ -35,6 +35,6 @@ models:
   - name: message_datetime
     data_type: timestamp
     description: The date and time the message was sent.
-  - name: extracted_at
+  - name: extracted_datetime
     data_type: timestamp
     description: The date and time the message was extracted from the Slack API. Part of combined primary key.

shared/projects/dbt/slack_analytics/models/staging/_models.yml

@@ -2,7 +2,8 @@ version: 2
 
 sources:
   - name: slack_analytics
+    database: ""
     schema: main
     tables:
       - name: channel_messages
-        identifier: src_channel_messages
+        identifier: channel_messages