add journey_id in activations table

treasure-data · Feb 19, 2024 · 30864b1 · 30864b1
1 parent 6a276c3
commit 30864b1
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 17 deletions.
diff --git a/scenarios/cdp_campaign_management/py_scripts/ingest_activations_queries.py b/scenarios/cdp_campaign_management/py_scripts/ingest_activations_queries.py
@@ -49,8 +49,8 @@ def run(
     columns_q = ['time','db_name','table_name','query']
 
     for idx, row in df_1.iterrows():
-
-        journey_table = 'journey_%s' % (row['journey_id'])
+        journey_id = row['journey_id']
+        journey_table = 'journey_%s' % (journey_id)
         syndication_id = row['syndication_id']
         activation_step_id = row['activation_step_id']
         activation_name = row['activation_name']
@@ -104,6 +104,7 @@ def run(
                 f",s1.activation_step_id",
                 f",s1.syndication_id",
                 f",'journeyActivationStep' AS activation_type",
+                f",{journey_id} AS journey_id",
                 f",s1.activation_name",
                 f",cv_name",
                 f",utm_campaign",

diff --git a/scenarios/cdp_campaign_management/queries/create_activations.sql b/scenarios/cdp_campaign_management/queries/create_activations.sql
@@ -4,6 +4,7 @@ CREATE TABLE IF NOT EXISTS ${dest_db}.${dest_table} (
     ,${user_id} varchar
     ,activation_step_id varchar
     ,syndication_id varchar
+    ,journey_id varchar
     ,activation_type varchar
     ,activation_name varchar
     ,cv_name varchar

diff --git a/scenarios/cdp_campaign_management/queries/ingest_master_activations.sql b/scenarios/cdp_campaign_management/queries/ingest_master_activations.sql
@@ -26,21 +26,21 @@ WITH tbl_act AS (
         ,journey_name
     FROM
     (
-    SELECT
-        elm
-        ,idx-1 AS stage_no
-        ,TD_TIME_PARSE(
-            JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.createdAt')
-            ,'${timezone}'
-        ) AS time
-        ,JSON_EXTRACT_SCALAR(elm, '$.name') AS stage_name
-        ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.audienceId') AS ps_id
-        ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.state') AS state
-        ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.createdAt') AS created_at
-        ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.updatedAt') AS updated_at
-        ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.name') AS journey_name
-    FROM ${td.monitoring.db.cdp_monitoring}.${td.monitoring.tables.journey_summary}
-    CROSS JOIN UNNEST(CAST(JSON_EXTRACT(JSON_PARSE(attributes), '$.journeyStages') AS ARRAY(JSON))) WITH ORDINALITY AS t(elm, idx)
+        SELECT
+            elm
+            ,idx-1 AS stage_no
+            ,TD_TIME_PARSE(
+                JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.createdAt')
+                ,'${timezone}'
+            ) AS time
+            ,JSON_EXTRACT_SCALAR(elm, '$.name') AS stage_name
+            ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.audienceId') AS ps_id
+            ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.state') AS state
+            ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.createdAt') AS created_at
+            ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.updatedAt') AS updated_at
+            ,JSON_EXTRACT_SCALAR(JSON_PARSE(attributes), '$.name') AS journey_name
+        FROM ${td.monitoring.db.cdp_monitoring}.${td.monitoring.tables.journey_summary}
+        CROSS JOIN UNNEST(CAST(JSON_EXTRACT(JSON_PARSE(attributes), '$.journeyStages') AS ARRAY(JSON))) WITH ORDINALITY AS t(elm, idx)
     )
     CROSS JOIN UNNEST(
         MAP_KEYS(CAST(JSON_EXTRACT(elm, '$.steps') AS MAP(VARCHAR, JSON)))