cief/dbt_cloud
1
0
Fork 0
mirror of https://gitlab.com/cief-data/dbt_cloud.git synced 2026-08-19 12:24:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate to new airbyte format

Browse Source
This commit is contained in:
CIEF ACC1
2023-11-03 08:39:06 +00:00
parent 30a08de7a0
commit ed72fe1a6e
1 changed files with 43 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/staging/googlesheet/stg_googlesheet__sales_exchange_lead_activations.sql
+43 -33
View File
@@ -1,49 +1,57 @@
-- IMPORT
WITH lead_activations AS (
SELECT * FROM {{ source('src_googlesheet_airbyte', '_airbyte_raw_leads_activations') }}
SELECT * FROM {{ source('src_googlesheet_airbyte', 'leads_activations') }}
),
-- LOGIC
parse_json_leads AS (
data_casting AS (
SELECT
_airbyte_ab_id AS sales_team_lead_activation_id,
parse_json(_airbyte_data) AS leads_json,
_airbyte_emitted_at AS _airbyte_emitted_datetime
_airbyte_raw_id AS _airbyte_id,
company_marking AS company_marking_id,
agent_name,
contact_number,
user_name AS contact_name,
email AS contact_email,
remark AS sales_team_remark,
sorce AS contact_method,
tag AS contact_status,
COALESCE(
TRY_TO_TIMESTAMP(register_date::STRING, 'DD-MM-YYYY'),
TRY_TO_TIMESTAMP(register_date::STRING, 'DD-MM-YYYY HH24:MI:SS')
) AS exchange_registered_datetime,
COALESCE(
TRY_TO_TIMESTAMP(user_first_reply_datetime::STRING, 'DD-MM-YYYY'),
TRY_TO_TIMESTAMP(user_first_reply_datetime::STRING, 'DD-MM-YYYY HH24:MI:SS')
) AS first_reply_datetime,
COALESCE(
TRY_TO_TIMESTAMP(contacted_datetime::STRING, 'DD-MM-YYYY'),
TRY_TO_TIMESTAMP(contacted_datetime::STRING, 'DD-MM-YYYY HH24:MI:SS')
) AS sales_team_contacted_datetime,
_airbyte_extracted_at AS _airbyte_extracted_datetime
FROM lead_activations
),
flatten_json AS (
generate_surrogate_key AS (
SELECT
sales_team_lead_activation_id,
leads_json['agent_name']::STRING AS agent_name,
leads_json['company_marking']::STRING AS company_marking_id,
leads_json['contact_number']::STRING AS contact_number,
leads_json['email']::STRING AS contact_email,
leads_json['sorce']::STRING AS contact_method,
leads_json['tag']::STRING AS contact_status,
leads_json['user_name']::STRING AS contact_name,
leads_json['remark']::STRING AS sales_team_remark,
COALESCE(
TRY_TO_TIMESTAMP(leads_json['contacted_datetime']::STRING, 'DD-MM-YYYY'),
TRY_TO_TIMESTAMP(leads_json['contacted_datetime']::STRING, 'DD-MM-YYYY HH24:MI:SS')
) AS sales_team_contacted_datetime,
COALESCE(
TRY_TO_TIMESTAMP(leads_json['register_date']::STRING, 'DD-MM-YYYY'),
TRY_TO_TIMESTAMP(leads_json['register_date']::STRING, 'DD-MM-YYYY HH24:MI:SS')
) AS exchange_registered_datetime,
COALESCE(
TRY_TO_TIMESTAMP(leads_json['user_first_reply_datetime']::STRING, 'DD-MM-YYYY'),
TRY_TO_TIMESTAMP(leads_json['user_first_reply_datetime']::STRING, 'DD-MM-YYYY HH24:MI:SS')
) AS first_reply_datetime,
*,
{{ dbt_utils.generate_surrogate_key(['sales_team_contacted_datetime', 'company_marking_id', 'agent_name']) }} AS sales_team_lead_activation_id,
_airbyte_emitted_datetime,
'{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
'{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
FROM data_casting
FROM parse_json_leads
),
-- FINAL
final__stg_googlesheet__sales_exchange_lead_activations AS (
@@ -68,11 +76,13 @@ final__stg_googlesheet__sales_exchange_lead_activations AS (
first_reply_datetime,
exchange_registered_datetime,
-- metadata
_airbyte_emitted_datetime,
-- metadata
_airbyte_id,
_airbyte_extracted_datetime,
_dbt_ran_datetime
FROM flatten_json
FROM generate_surrogate_key
)
SELECT * FROM final__stg_googlesheet__sales_exchange_lead_activations