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

Merge branch 'lee_address_geoloc' into 'main'

Browse Source 
Lee address geoloc

See merge request CIEFWorldwideSdnBhd/dbt_cloud!3
This commit is contained in:
Yam ZhengLim
2023-06-22 09:14:08 +00:00
parent c313097ac6 f21902d575
commit f2c8057886
11 changed files with 866 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2 -2
View File
@@ -1,7 +1,7 @@
# DBT governance
Thid markdown document will specify the rules that need to follow by every developer.
This markdown document will specify the rules that need to follow by every developer.
<br>
Short naming converntion lists that need to follow
Short naming convention lists that need to follow
| model_type Shortcut | Full name |
| ------------------- | ------ |
dbt_project.yml
+3 -1
View File
@@ -104,4 +104,6 @@ tests:
+severity: warn
vars:
mask_data_pattern: "('*MASK*', '****')"
mask_data_pattern_for_udf: "[*MASK*, ****]"
macros/_macro__docs.md
+362
View File
@@ -0,0 +1,362 @@
{% docs macro_address_to_coordinates %}
# Procedure to define E-UDF in GCP and calling the function in DBT
## 1.0 Information required to create E-UDF
As you complete the tasks to create an external function in the Google Cloud Console, you are required to record specific values (e.g. Cloud Function Trigger URL) during each step in the process. Whereby, the values you entered, will be required in subsequent steps.
To facilitate recording/tracking of this information, we have provided a template with fields for each of the required values. Save the template in a text editor for easier referencing.
Reference Source: [Snowflake Planning](https://docs.snowflake.com/en/sql-reference/external-functions-creating-gcp-planning)
```
=================== Tracking Worksehet: Google Cloud Console ===================
Step 1: Cloud Function (Remote Service) Info
Cloud Function Trigger URL: ____________________________________________________
Step 2: API Config File Info
Path Suffix: ___________________________________________________________________
Configuration File Name: _______________________________________________________
Step 3: API Gateway (Proxy Service) Info
Managed Service Identifier: ____________________________________________________
Gateway Base URL : _____________________________________________________________
Steps 4: API Integration & External Function Info
API Integration Name: __________________________________________________________
API_GCP_SERVICE_ACCOUNT: _______________________________________________________
External Function Name: ________________________________________________________
Step 5: Security Info
Security Definition Name: ______________________________________________________
```
<br>
## 2.0 Config E-UDF in GCP & Snowflake
### Step 1. Create the Cloud Function in GCP
Reference Source: [Snowflake Create the Remote Service](https://docs.snowflake.com/en/sql-reference/external-functions-creating-gcp-ui-remote-service)
1. Create function from **GCP Cloud functions service**
🔍 GCP Search: Cloud Function
2. Configuration to set up the cloud function are as followed:
- **Basics**
- Environment: 1st gen
- Function name: `<function_name_of_your_choice>`
- Region: us-central1
- **Trigger**
- Trigger type: HTTP
- Authentication: Require authentication
- **Runtime**
- Memory allocated: 256MB
- Timeout: 240
3. Save the configuration.
4. Proceed by clicking next and work on inputing the function program in the window.
5. Deploy the function when ready.
6. Under Google Cloud Function UI, select the deployed function and select the trigger tab.
7. Copy the trigger URL of the function and record into the template field <mark>Step 1: Cloud function trigger URL</mark>. For example: `https://us-central1-data-management-123456.cloudfunctions.net/address_to_geo_coordinate`
<br>
### Step 2. Create Proxy Service in GCP
Reference Source: [Snowflake Create the Proxy Service](https://docs.snowflake.com/en/sql-reference/external-functions-creating-gcp-ui-proxy-service)
1. On your local file system (desktop), create a YAML config file with the `unique name` of your choice and end with the extension `.yml`. For example, `config_address_to_geo_coordinate.yml`
2. Copy the following configuration template into the YAML file.
```yml
swagger: '2.0'
info:
title: API Gateway config for Snowflake external function.
description: This configuration file connects the API Gateway resource to the remote service (Cloud Function).
version: 1.0.0
schemes:
- https
produces:
- application/json
paths:
/<PATH>:
post:
summary: Echo the input.
operationId: echo
x-google-backend:
address: <HTTP ENDPOINT TO ROUTE REQUEST TO>
protocol: h2
responses:
'200':
description: <DESCRIPTION>
schema:
type: string
```
3. Under paths, replace `<PATH>` with a unique name. For example, `func-address-to-geo`. Do not remove the forward slash.
4. Save the path name into the template field <mark>Step 2: Path Suffix</mark>. For example, `func-address-to-geo`
5. Replace `<HTTP ENDPOINT TO ROUTE REQUEST TO>` with the value previously recorded in the template field <mark>Step 1: Cloud Function Trigger URL</mark> ***The URL should not be enclosed in "quotation marks"***. For example, `https://us-central1-data-management-123456.cloudfunctions.net/address_to_geo_coordinate`
6. Save the file and record the yaml file name into the template field <mark>Step 3: Configuration File Name</mark>. For example, `config_address_to_geo_coordinate.yml`
7. Search for Google Cloud API Gateway service in GCP.
🔍 GCP Search: API Gateway
8. Create a Gateway and configure as followed:
- **API**
- Select an API: Create new API
- Display Name: `<display_name_of_your_choice>`
- API ID: `<API_ID_of_your_choice>`
- Region: us-central1
- **API Config**
- Select a Config: Create new API config
- Upload an API Spec: Upload the config yaml file as configured in previous step from your local machine.
- Display Name: `<display_name_of_your_choice>`
- Select a Service Account: App Engine default service account
- **Gateway details**
- Display Name: `<display_name_of_your_choice>`
- Location: us-central1
9. Click create gateway and wait several minutes for it to create.
10. In the API gateway page, copy the value from the managed service for the newly created API. Record this value into the template field <mark>Step 3: Managed Service Identifier</mark>. For example, `api-address-to-geo-coordinates-external-function-123456abcde.apigateway.data-management-123456.cloud.goog`
11. In the API gateway page, select the API and click on the Gateways tab.
12. Copy Gateway URL and record this information into the template file <mark>Step 3: Gateway Base URL</mark>. For example, `https://gateway-address-to-coordinates-12345abc.uc.gateway.dev`
### Step 3. Create API Integration in Snowflake
Reference Source: [Snowflake Create the API Integration](https://docs.snowflake.com/en/sql-reference/external-functions-creating-gcp-common-api-integration)
1. Open a new Snowflake sql worksheet.
2. Ensure current snowflake account has accountadmin privileges.
3. Enter following sql statement template:
```sql
create or replace api integration <integration_name>
api_provider = google_api_gateway
google_audience = '<google_audience_claim>'
api_allowed_prefixes = ('<url>')
enabled = true;
```
4. Replace `<integration_name>` with `<integration_name_of_your_choice>` and record the newly defined name into the template field <mark>Step 4: API Integration Name</mark>. For example, `gcp_api_integration_address_to_coordinates`
5. Replace `<google_audience_claim>` with value previously recorded from template field <mark>Step 3: Managed Service Identifier</mark>
6. Execute the statement.
7. Run the following sql statement:
```sql
describe integration <newly_defined_integration_name_in_previous_step>;
```
8. Record `API_GCP_SERVICE_ACCOUNT` value obtained from the execution into the template field <mark>Step 4: API_GCP_SERVICE_ACCOUNT</mark>. For example, `abc1234567@gcpuscentral1-abc1.iam.gserviceaccount.com`
### Step 4. Create External Function in Snowflake
Reference Source: [Snowflake Create the External Function](https://docs.snowflake.com/en/sql-reference/external-functions-creating-gcp-common-ext-function)
1. In the snowflake sql worksheet, enter the following statement:
```sql
create or replace external function <external_function_name>(<parameters>)
returns variant
api_integration = <api_integration_name>
as '<function_url>';
```
2. Replace `<external_function_name>` with a unique function name of your choice. For example, `echo_address_to_coordinates`
3. Replace `<parameters>` based on your external function configuration. For example, `id VARCHAR, full_address VARCHAR`
4. Replace `<api_integration_name>` with the value previously recorded from template field <mark>Step 4: API Integration Name</mark>
5. Replace `<function_URL>` with the value previously recorded from template field <mark>Step 3: Gateway Base URL</mark> and <mark>Step 2: Path Suffix</mark>. Combine the string by a forward slash ( / ) between them. For example: `'https://gateway-address-to-coordinates-12345abc.uc.gateway.dev/func-address-to-geo'`
6. Execute the statement.
<br>
## Security Policy
<br>
## 3.0 Calling E-UDF Function in Snowflake
1. Calling E-UDF in the schema by the name of the E-UDF defined previously. For example:
```sql
SELECT echo_address_to_coordinates(full_address) FROM address;
```
2. If the E-UDF is not stored in the same directory as your current working directory, you need to specify the database and schema name of the E-UDF. For example:
```sql
SELECT dev_data_warehouse.dev_schema.echo_address_to_coordinates(full_address) FROM address;
```
<br>
## 4.0 Defining and calling macro and global variable for the E-UDF in DBT
### Step 1: Defining macro for geocoding address
This macro converts an address string into a set of geographical coordinates. The macro will filter the rows, excluding masked data rows based on data mask pattern defined in a global variable. It accepts five input parameters representing address components and returns the parsed JSON response from the E-UDF. This set of instructions are for used in dbt.
1. Define a macro in the /macros folder directing towards the E-UDF. Remember to specify the database and schema name that store the E-UDF.
The name of the macro utilized here is address_to_coordinates()
Parameters:
address_line_one: VARCHAR or STRING - Represents ADDRESS_LINE_ONE.
address_line_two: VARCHAR or STRING - Represents ADDRESS_LINE_TWO.
postcode : VARCHAR or STRING - Represents POSTCODE.
district_name : VARCHAR or STRING - Represents DISTRICT_NAME.
state_name : VARCHAR or STRING - Represents STATE_NAME.
```sql
-- Remeber to add { } braces at the start and end of the jinja statement %
-- Add double { } on the following locations
-- Around the global variable var('mask_data_pattern')
%- macro address_to_coordinates(address_line_one, address_line_two, postcode, district_name, state_name) -%
CASE
WHEN
( address_line_one IN var('mask_data_pattern') OR address_line_two in var('mask_data_pattern')
OR
address_line_one IS NULL AND address_line_two IS NULL )
THEN NULL
ELSE
PARSE_JSON(global_function.eudf.echo_address_to_coordinates(address_line_one, address_line_two, postcode, district_name, state_name))
END
%- endmacro -%
```
### Step 2: Defining global variable
A global variable can be defined in `dbt_project.yml` file. The variable here is called `mask_data_pattern` which is to store a list of string patterns used for masking data.
```yml
vars:
mask_data_pattern: ('*MASK*', '****')
```
### Step 3: Implement incremental model
Include a materialize configuration block to the model to specify the model as 'incremental' and input the unique_key value(s).
- Snowflake by default uses the merge method for incremental strategy. Thus this model adopts the default method.
- The unique_key must not have null and duplicates. It is suggested to use the primary key of the table instead.
- One or more unique_key can be selected.
- Add double { } to wrap around the configuration block
```sql
config(
materialized='incremental',
unique_key=['COMPANY_ID', 'COMPANY_ADDRESS_ID']
)
```
Inlcude a filter block using is_incremental() function for the model to define how to update the model.
The following example uses the 'UPDATED_DATETIME' as the filter to identify new rows to undergo transformation.
- Add double { } to wrap around the following locations:
- ref('test_address__mock_data')
- this
- Remeber to add { } braces at the start and end of the jinja statement %
```sql
WITH events as (
SELECT * FROM ref('test_address__mock_data')
% if is_incremental() %
WHERE UPDATED_DATETIME > (SELECT max(UPDATED_DATETIME) FROM this )
%- endif %
),
```
### Step 4: Calling the macro
To call the defined macro, the following syntax can be utilized, take note to **include** the **double curly braces**:
- In addition, underlying values can be derived from the results of the macro. You can access the values in the list by referring to the name of the key.
- Shown in the second code block for example, the latitude and longitude of the location is retrieved and stored in a new column.
```sql
-- Add double { } on the following locations
-- Around the E-UDF address_to_coordinates()
-- Around the ref('test_address__mock_data')
WITH geocoded_address AS (
SELECT
*,
address_to_coordinates( 'ADDRESS_LINE_ONE', 'ADDRESS_LINE_TWO', 'POSTCODE', 'DISTRICT_NAME', 'STATE_NAME' ) AS FUNCTION_JSON_RETURN
FROM events
),
json_data_extraction AS (
SELECT
FUNCTION_JSON_RETURN:result:geometry:location:lat::float AS LATITUDE,
FUNCTION_JSON_RETURN:result:geometry:location:lng::float AS LONGITUDE,
FUNCTION_JSON_RETURN:geocoding_api_returns AS API_GEOCODING_JSON_RETURN,
FUNCTION_JSON_RETURN:result AS API_PLACE_DETAILS_JSON_RETURN,
*
FROM geocoded_address
)
```
## 5.0 Local Variables
Several local variables are defined using Jinja templating:
- `days_of_week`: A list of strings representing the days of the week.
- `masked_address_sql`: A SQL condition statement to check if the address line is masked. This variable also refers to the global variable 'mask_data_pattern'.
- `dummy_value`: A dummy value used for placeholders.
- `dummy_long_string`: A long dummy string used for placeholders.
- `dummy_short_string`: A short dummy string used for placeholders.
- `dummy_phone`: A dummy phone number string used for placeholders.
- `dummy_operating_hours`: A dummy operating hours string used for placeholders.
- `dummy_website`: A dummy website URL string used for placeholders.
## 6.0 CTEs (Common Table Expressions)
`company_addresses_with_row_num` <br>
This CTE assigns a row number to each row in the company_addresses table and filters the data based on the incremental update condition if the job is incremental.
`company_addressess_join_companies_countries_states_districts` <br>
This CTE joins the company_addresses_with_row_num CTE with the countries, states, and districts tables to form a fuller table.
`batched_data_batch` <br>
This CTE splits the data rows into batches for processing. Each batch contains up to 10 rows. This is to avoid timeout and resource exhaustion experienced in snowflake caused by concurrency.
`address_enrich_using_google_api_batch` <br>
This CTE enriches the address data by using the address_to_coordinates E-UDF, which calls the Google Maps API to obtain geocoding and location information for each address in the batch.
`union_all_batch_address_enriched` <br>
This CTE combines the results from all the batches into a single table.
`operation_hours_json` <br>
This CTE extracts the operating hour range for each day of the week from the JSON data returned by the API.
`operation_hour_range_extraction` <br>
This CTE further processes the extracted operating hour ranges and removes the day prefixes.
`operation_hour_to_24_hour_transform` <br>
This CTE converts the operating hour ranges to a 24-hour format.
`special_single_operation_hour_to_24_hour_transform` <br>
This CTE handles cases where there are only single operating hours within a day but do not have an AM/PM indicator and converts them to a 24-hour format.
`special_double_operation_hour_to_24_hour_transform` <br>
This CTE handles cases where there are two operating hours within a day and converts them to a 24-hour format.
`concatenate_operation_hour` <br>
This CTE concatenates the first and second transformed time ranges for each day of the week. For scenarios with only single operating hours within a day, the original time range will be returned.
`google_place_api_details_extraction` <br>
This CTE further extracts potentially valuable information returned by the API and flatten them.
`masked_data_transformation` <br>
This CTE applies transformations to the selected metrics and dimensions. It replaces sensitive information with dummy values if the address is masked.
{% enddocs %}
macros/_macro__models.yml
+21
View File
@@ -0,0 +1,21 @@
version: 2
macros:
- name: address_to_coordinates
description: '{{ doc("macro_address_to_coordinates") }}'
arguments:
- name: address_line_one
type: string
description: The first line of the address, typically representing the primary address information such as street number, name, and building name.
- name: address_line_two
type: string
description: The second line of the address, used to provide additional address details if needed, such as apartment number, suite number, or floor.
- name: postcode
type: integer
description: The postal code or ZIP code of the address, which identifies a specific geographic area.
- name: district_name
type: string
description: The name of the district or neighborhood where the address is located.
- name: state_name
type: string
description: The name of the state or region where the address is situated.
macros/address_to_coordinates.sql
+22
View File
@@ -0,0 +1,22 @@
/*
A macro to convert an address string into a set of geo coordinates.
It accepts five input parameters representing address components and returns the parsed JSON response from the E-UDF.
Parameters:
- address_line_one: VARCHAR or STRING - First line of address
- address_line_two: VARCHAR or STRING - Second line of address
- postcode: VARCHAR or STRING - Postcode number
- district_name: VARCHAR or STRING - Name of district
- state_name: VARCHAR or STRING - Name of state
Returns:
- VARIANT - A parsed JSON response from the E-UDF, representing metadata from geocoding the address.
*/
{%- macro address_to_coordinates(masking_pattern, address_line_one, address_line_two, postcode, district_name, state_name) -%}
PARSE_JSON(global_function.eudf.echo_address_to_coordinates( '{{var('mask_data_pattern_for_udf')}}', address_line_one, address_line_two, postcode, district_name, state_name ))
{%- endmacro -%}
models/marts/reporting/rep_exchange__daily_orders.sql
+4
View File
@@ -51,6 +51,8 @@ fct_and_dim_joins AS (
companies.district_name AS company_district_name,
companies.postcode AS company_postcode,
companies.has_wallet AS company_has_wallet,
companies.latitude AS company_latitude,
companies.longitude AS company_longitude,
bookings.user_id,
bookings.bank_id,
@@ -191,6 +193,8 @@ final__rep_exchange__daily_orders AS (
company_state_name,
company_district_name,
company_postcode,
company_latitude,
company_longitude,
company_has_wallet,
cost_transaction_type,
cost_payment_method,
models/marts/warehouse/_warehouse_models.yml
+1 -1
View File
@@ -28,7 +28,7 @@ models:
- unique
- not_null
- name: fct_exchange__costs
- name: fct_exchange__transaction_costs
description: exchange transaction_cost
columns:
- name: transaction_cost_id
models/marts/warehouse/dim_exchange__companies.sql
+36
View File
@@ -97,6 +97,24 @@ companies_rename_and_join_table AS (
company_billing_addresses.postcode,
company_billing_addresses.address_line_one,
company_billing_addresses.address_line_two,
company_billing_addresses.google_place_id,
company_billing_addresses.phone_number AS google_api_extracted_phone_number,
company_billing_addresses.google_returned_address,
company_billing_addresses.location_type,
company_billing_addresses.operational_status,
company_billing_addresses.latitude,
company_billing_addresses.longitude,
company_billing_addresses.location_website,
company_billing_addresses.location_overall_rating,
company_billing_addresses.location_number_of_reviews,
company_billing_addresses.monday_operating_hours,
company_billing_addresses.tuesday_operating_hours,
company_billing_addresses.wednesday_operating_hours,
company_billing_addresses.thursday_operating_hours,
company_billing_addresses.friday_operating_hours,
company_billing_addresses.saturday_operating_hours,
company_billing_addresses.sunday_operating_hours,
company_billing_addresses.json_place_details_full_return,
company_wallets.wallet_id,
IFF(company_wallets.wallet_id IS NOT NULL, 1, 0) AS has_wallet,
@@ -152,6 +170,7 @@ final__dim_exchange__companies AS (
wallet_id,
wallet_currency_id,
identity_document_id,
google_place_id,
-- dimensions
name,
@@ -176,6 +195,23 @@ final__dim_exchange__companies AS (
identity_document_type,
identity_reference,
identity_status,
google_returned_address,
google_api_extracted_phone_number,
location_type,
operational_status,
latitude,
longitude,
location_website,
location_overall_rating,
location_number_of_reviews,
monday_operating_hours,
tuesday_operating_hours,
wednesday_operating_hours,
thursday_operating_hours,
friday_operating_hours,
saturday_operating_hours,
sunday_operating_hours,
json_place_details_full_return,
-- measures
wallet_balance,
models/staging/exchange/_exchange__models.yml
+3 -1
View File
@@ -23,4 +23,6 @@ models:
description: Bridge between `companies` and `users`
- name: stg_exchange__company_addresses
description: Company addresses that input when register
description: Company addresses that input when register
config:
tags: ['incremental']
models/staging/exchange/stg_exchange__company_addresses.sql
+408 -20
View File
@@ -1,3 +1,30 @@
-- A flag indicator to determine whether to run a full_load or an incremental_load
-- If user is a developer, a full load will be executed.
-- If the target schema is set to 'prod', only an incremental load will be performed.
{% if target.name == 'prod' %}{% set one_month_incremental_flag = true %}{% else %}{% set one_month_incremental_flag = false %}{% endif %}
-- MODEL CONFIGURATION BLOCK
{{
config(
materialized='incremental',
unique_key=['company_id', 'company_address_id']
)
}}
-- LOCAL VARIABLES
{%- set days_of_week = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] %}
{%- set masked_address_sql = "(address_line_one IN " + var('mask_data_pattern') + " OR address_line_two IN " + var('mask_data_pattern') + " OR address_line_one IS NULL AND address_line_two IS NULL)" %}
{%- set dummy_value = -999 %}
{%- set dummy_long_string = "xxxxxxxxxxxxxxxxxx" %}
{%- set dummy_short_string = "xxxxxx" %}
{%- set dummy_phone = "xxx-xxxxxxxxxx" %}
{%- set dummy_operating_hours = '99:99-99:99' %}
{%- set dummy_website = 'https:www://xxxxxxxxxx.com' %}
-- IMPORTS
WITH company_addresses AS (
SELECT * FROM {{ ref('base_exchange__company_addresses') }}
@@ -15,43 +42,385 @@ districts AS (
SELECT * FROM {{ ref('base_exchange__districts') }}
),
-- LOGIC
company_addresses_with_row_num AS (
SELECT
*,
ROW_NUMBER() OVER (ORDER BY company_address_id) AS row_num
FROM company_addresses
{% if is_incremental() %}
WHERE updated_datetime > ( SELECT max(updated_datetime) FROM {{ this }} )
{%- endif %}
),
company_addressess_join_companies_countries_states_districts AS (
SELECT
company_addresses.company_address_id,
company_addresses.company_id,
company_addresses_with_row_num.row_num,
company_addresses.country_id,
company_addresses_with_row_num.company_address_id,
company_addresses_with_row_num.company_id,
company_addresses_with_row_num.country_id,
countries.name AS country_name,
countries.short_code AS country_short_code,
countries.phone_code AS country_phone_code,
company_addresses.state_id,
company_addresses_with_row_num.state_id,
states.name AS state_name,
company_addresses.district_id,
company_addresses_with_row_num.district_id,
districts.name AS district_name,
company_addresses.postcode,
company_addresses.address_line_one,
company_addresses.address_line_two,
company_addresses.is_billing_address,
company_addresses.deleted_datetime,
company_addresses.created_datetime,
company_addresses.updated_datetime,
'{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
FROM company_addresses
company_addresses_with_row_num.postcode,
company_addresses_with_row_num.address_line_one,
company_addresses_with_row_num.address_line_two,
company_addresses_with_row_num.is_billing_address,
company_addresses_with_row_num.deleted_datetime,
company_addresses_with_row_num.created_datetime,
company_addresses_with_row_num.updated_datetime
FROM company_addresses_with_row_num
LEFT JOIN countries
ON (company_addresses.country_id = countries.country_id)
ON (company_addresses_with_row_num.country_id = countries.country_id)
LEFT JOIN states
ON (company_addresses.state_id = states.state_id)
ON (company_addresses_with_row_num.state_id = states.state_id)
LEFT JOIN districts
ON (company_addresses.district_id = districts.district_id)
ON (company_addresses_with_row_num.district_id = districts.district_id)
),
-- Setting total rows to be iterated
{% if is_incremental() and one_month_incremental_flag %}
-- Get previous 1 month data from current time
{% set estimate_incremental_rows = dbt_utils.get_column_values(
table=ref('base_exchange__company_addresses'),
where='updated_datetime > ADD_MONTHS(CURRENT_DATE(), -1)',
column='company_address_id') %}
{%- set total_rows = estimate_incremental_rows|length %}
{% else %}
{%- set company_address_ids = dbt_utils.get_column_values(table=ref('base_exchange__company_addresses'), column='company_address_id') %}
{%- set total_rows = company_address_ids|length %}
{% endif %}
-- Formulating row batches for batch processing
-- Each batch is recommended to limit to 10 rows to avoid timeout
{%- set batch_size = 10 %}
{%- set num_batches = (total_rows // batch_size) + 1 %}
{%- set current_batch = 1 %}
{%- for batch in range(1, num_batches + 1) %}
{%- set start_row = (batch - 1) * batch_size + 1 %}
{%- set end_row = start_row + batch_size - 1 %}
batched_data_{{batch}} AS (
SELECT * FROM company_addressess_join_companies_countries_states_districts
WHERE row_num >= {{ start_row }} AND row_num <= {{ end_row }}
),
{%- endfor %}
{%- for batch in range(1, num_batches + 1) %}
address_enrich_using_google_api_{{batch}} AS (
SELECT
*,
{{ address_to_coordinates( 'address_line_one', 'address_line_two', 'postcode', 'district_name', 'state_name' ) }} AS json_function_return
FROM batched_data_{{batch}}
),
{%- endfor %}
union_all_batch_address_enriched AS (
{%- for batch in range(1, num_batches + 1) %}
SELECT * FROM address_enrich_using_google_api_{{batch}}
{%- if not loop.last %} UNION{% endif -%}
{% endfor %}
),
operation_hours_json AS (
SELECT
*,
-- Extract operating hour range for each day of the week and perform unicode replacement
REGEXP_REPLACE(REGEXP_REPLACE(json_function_return['output_result']['result']['opening_hours']['weekday_text'][0]::string, '[\u2013\u002d]', '-'), '[\u202F\u2009]', ' ') AS str_monday_operating_hour,
REGEXP_REPLACE(REGEXP_REPLACE(json_function_return['output_result']['result']['opening_hours']['weekday_text'][1]::string, '[\u2013\u002d]', '-'), '[\u202F\u2009]', ' ') AS str_tuesday_operating_hour,
REGEXP_REPLACE(REGEXP_REPLACE(json_function_return['output_result']['result']['opening_hours']['weekday_text'][2]::string, '[\u2013\u002d]', '-'), '[\u202F\u2009]', ' ') AS str_wednesday_operating_hour,
REGEXP_REPLACE(REGEXP_REPLACE(json_function_return['output_result']['result']['opening_hours']['weekday_text'][3]::string, '[\u2013\u002d]', '-'), '[\u202F\u2009]', ' ') AS str_thursday_operating_hour,
REGEXP_REPLACE(REGEXP_REPLACE(json_function_return['output_result']['result']['opening_hours']['weekday_text'][4]::string, '[\u2013\u002d]', '-'), '[\u202F\u2009]', ' ') AS str_friday_operating_hour,
REGEXP_REPLACE(REGEXP_REPLACE(json_function_return['output_result']['result']['opening_hours']['weekday_text'][5]::string, '[\u2013\u002d]', '-'), '[\u202F\u2009]', ' ') AS str_saturday_operating_hour,
REGEXP_REPLACE(REGEXP_REPLACE(json_function_return['output_result']['result']['opening_hours']['weekday_text'][6]::string, '[\u2013\u002d]', '-'), '[\u202F\u2009]', ' ') AS str_sunday_operating_hour
FROM union_all_batch_address_enriched
),
operation_hour_range_extraction AS (
SELECT
*,
{% for day_of_week in days_of_week %}
-- Extract the time range for each day of the week with day prefix removed
REGEXP_REPLACE(str_{{day_of_week}}_operating_hour, '^[a-zA-Z]+: ', '') AS {{day_of_week}}_time_range
{% if not loop.last %}, {% endif %}
{% endfor %}
FROM operation_hours_json
),
operation_hour_to_24_hour_transform AS (
SELECT
*,
{%- for day_of_week in days_of_week %}
CASE
-- Convert time range to 24-hour format for standard operating hour format. "9:00 AM - 10:00 PM"
WHEN REGEXP_LIKE({{day_of_week}}_time_range, '^[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM][ ]?-[ ]?[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM]$') THEN
CONCAT
(
TO_CHAR(TO_TIME(SPLIT_PART({{day_of_week}}_time_range, ' - ', 1)), 'HH24:MI'),
' - ',
TO_CHAR(TO_TIME(SPLIT_PART({{day_of_week}}_time_range, ' - ', 2)), 'HH24:MI')
)
ELSE {{day_of_week}}_time_range
END AS transformed_{{day_of_week}}_operating_hours
{% if not loop.last%}, {% endif %}
{% endfor %}
FROM operation_hour_range_extraction
),
special_single_operation_hour_to_24_hour_transform AS (
SELECT
*,
{%- for day_of_week in days_of_week %}
-- Convert time range to 24-hour format for cases where the AM/PM indicator is missing in the opening hour
CASE WHEN REGEXP_LIKE(transformed_{{day_of_week}}_operating_hours, '^[0-9]{1,2}:[0-9]{2}[ ]?-[ ]?[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM]$') THEN
CONCAT(TO_CHAR(TO_TIME(
CASE
WHEN TO_TIME(SPLIT_PART(transformed_{{day_of_week}}_operating_hours, ' - ', 2)) > '12:00 PM' THEN
SPLIT_PART(transformed_{{day_of_week}}_operating_hours, ' - ', 1) || ' PM'
ELSE
SPLIT_PART(transformed_{{day_of_week}}_operating_hours, ' - ', 1)
END
),'HH24:MI'),
' - ',
TO_CHAR(TO_TIME(SPLIT_PART(transformed_{{day_of_week}}_operating_hours, ' - ', 2)), 'HH24:MI')
)
ELSE transformed_{{day_of_week}}_operating_hours
END AS transformed_{{day_of_week}}_time_range
{% if not loop.last%}, {% endif %}
{% endfor %}
FROM operation_hour_to_24_hour_transform
),
-- This block handles specifically scenarios that has 2 operational hours in a single day "9:00 AM - 3:00 PM, 5:00 PM - 10:00 PM"
special_double_operation_hour_to_24_hour_transform AS (
SELECT
*,
{%- for day_of_week in days_of_week %}
CASE
-- First operating hour
-- Convert time range to 24-hour format for cases where the AM/PM indicator is missing in the operating hour
WHEN REGEXP_LIKE(operation_hour_split.{{day_of_week}}_first_operation_hour, '^[0-9]{1,2}:[0-9]{2}[ ]?-[ ]?[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM]$') THEN
CONCAT(TO_CHAR(TO_TIME(
CASE
WHEN TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 2)) > '12:00 PM' THEN
SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 1) || ' PM'
ELSE
SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 1)
END
),'HH24:MI'),
' - ',
TO_CHAR(TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 2)), 'HH24:MI')
)
-- Convert time range to 24-hour format for standard operating hour format
WHEN REGEXP_LIKE(operation_hour_split.{{day_of_week}}_first_operation_hour, '^[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM][ ]?-[ ]?[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM]$') THEN
CONCAT
(
TO_CHAR(TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 1)), 'HH24:MI'),
' - ',
TO_CHAR(TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 2)), 'HH24:MI')
)
ELSE operation_hour_split.{{day_of_week}}_first_operation_hour
END AS {{day_of_week}}_transformed_first_time_range,
-- Second operating hour
-- Convert time range to 24-hour format for cases where the AM/PM indicator is missing in the operating hour
CASE WHEN REGEXP_LIKE(operation_hour_split.{{day_of_week}}_second_operation_hour, '^[0-9]{1,2}:[0-9]{2}[ ]?-[ ]?[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM]$') THEN
CONCAT(TO_CHAR(TO_TIME(
CASE
WHEN TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_second_operation_hour, ' - ', 2)) > '12:00 PM' THEN
SPLIT_PART(operation_hour_split.{{day_of_week}}_second_operation_hour, ' - ', 1) || ' PM'
ELSE
SPLIT_PART(operation_hour_split.{{day_of_week}}_second_operation_hour, ' - ', 1)
END
),'HH24:MI'),
' - ',
TO_CHAR(TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_second_operation_hour, ' - ', 2)), 'HH24:MI')
)
--- Convert time range to 24-hour format for standard operating hour format
WHEN REGEXP_LIKE(operation_hour_split.{{day_of_week}}_first_operation_hour, '^[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM][ ]?-[ ]?[0-9]{1,2}:[0-9]{2}[ ]?[aApP][mM]$') THEN
CONCAT
(
TO_CHAR(TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 1)), 'HH24:MI'),
' - ',
TO_CHAR(TO_TIME(SPLIT_PART(operation_hour_split.{{day_of_week}}_first_operation_hour, ' - ', 2)), 'HH24:MI')
)
ELSE operation_hour_split.{{day_of_week}}_second_operation_hour
END AS {{day_of_week}}_transformed_second_time_range
{% if not loop.last%}, {% endif %}
{% endfor %}
FROM
-- Subquery to extract first and second operational hour of the day
( SELECT
*,
{%- for day_of_week in days_of_week %}
SPLIT_PART(transformed_{{day_of_week}}_time_range, ', ', 1) AS {{day_of_week}}_first_operation_hour,
SPLIT_PART(transformed_{{day_of_week}}_time_range, ', ', 2) AS {{day_of_week}}_second_operation_hour
{% if not loop.last%}, {% endif %}
{% endfor %}
FROM special_single_operation_hour_to_24_hour_transform
) AS operation_hour_split
),
concatenate_operation_hour AS (
SELECT
*,
{%- for day_of_week in days_of_week %}
CASE
WHEN {{day_of_week}}_transformed_second_time_range <> '' THEN
CONCAT({{day_of_week}}_transformed_first_time_range, ', ', {{day_of_week}}_transformed_second_time_range)
ELSE {{day_of_week}}_transformed_first_time_range
END AS {{day_of_week}}_final_operating_hours
{% if not loop.last%}, {% endif %}
{% endfor %}
FROM special_double_operation_hour_to_24_hour_transform
),
google_place_api_details_extraction AS (
SELECT
*,
json_function_return['output_place_id']['place_id_value']::string AS google_place_id,
json_function_return AS json_place_details_full_return,
json_function_return['function_run_datetime']::datetime AS _function_ran_datetime,
json_function_return['output_result']['result']['formatted_address']::string AS google_returned_address,
json_function_return['output_result']['result']['formatted_phone_number']::string AS phone_number,
json_function_return['output_result']['result']['types']::string AS location_type,
json_function_return['output_result']['result']['business_status']::string AS operational_status,
json_function_return['output_result']['result']['geometry']['location']['lat']::float AS latitude,
json_function_return['output_result']['result']['geometry']['location']['lng']::float AS longitude,
json_function_return['output_result']['result']['website']::string AS location_website,
json_function_return['output_result']['result']['rating']::float AS location_overall_rating,
json_function_return['output_result']['result']['user_ratings_total']::integer AS location_number_of_reviews
FROM concatenate_operation_hour
),
masked_data_transformation AS (
SELECT
company_address_id,
company_id,
country_id,
state_id,
district_id,
CASE
WHEN {{ masked_address_sql }} THEN '{{ dummy_long_string }}'
ELSE google_place_id
END AS google_place_id,
-- dimensions
country_name,
country_short_code,
country_phone_code,
state_name,
district_name,
postcode,
address_line_one,
address_line_two,
is_billing_address,
CASE
WHEN {{ masked_address_sql }} THEN '{{ dummy_short_string }}'
ELSE google_returned_address
END AS google_returned_address,
CASE
WHEN {{ masked_address_sql }} THEN '{{ dummy_phone }}'
ELSE phone_number
END AS phone_number,
CASE
WHEN {{ masked_address_sql }} THEN '[' || '{{ dummy_short_string }}' || ']'
ELSE location_type
END AS location_type,
CASE
WHEN {{ masked_address_sql }} THEN '{{ dummy_short_string }}'
ELSE operational_status
END AS operational_status,
CASE
WHEN {{ masked_address_sql }} THEN {{ dummy_value }}
ELSE latitude
END AS latitude,
CASE
WHEN {{ masked_address_sql }} THEN {{ dummy_value }}
ELSE longitude
END AS longitude,
CASE
WHEN {{ masked_address_sql }} THEN '{{ dummy_website }}'
ELSE location_website
END AS location_website,
CASE
WHEN {{ masked_address_sql }} THEN {{ dummy_value }}
ELSE location_overall_rating
END AS location_overall_rating,
CASE
WHEN {{ masked_address_sql }} THEN {{ dummy_value }}
ELSE location_number_of_reviews
END AS location_number_of_reviews,
{%- for day_of_week in days_of_week %}
CASE
WHEN {{ masked_address_sql }} THEN '{{ dummy_operating_hours }}'
ELSE {{day_of_week}}_final_operating_hours
END AS {{day_of_week}}_operating_hours,
{% endfor %}
CASE
WHEN {{ masked_address_sql }} THEN '{"' || '{{dummy_long_string}}' || '"}'
ELSE json_place_details_full_return
END AS json_place_details_full_return,
-- date/times
deleted_datetime,
created_datetime,
updated_datetime,
-- metadata
_function_ran_datetime,
'{{ modules.datetime.datetime.now(modules.pytz.timezone("Asia/Kuala_Lumpur")) }}' AS _dbt_ran_datetime
FROM google_place_api_details_extraction
),
-- FINAL
@@ -64,6 +433,7 @@ final__stg_exchange__company_addresses AS (
country_id,
state_id,
district_id,
google_place_id,
-- dimensions
country_name,
@@ -75,6 +445,23 @@ final__stg_exchange__company_addresses AS (
address_line_one,
address_line_two,
is_billing_address,
google_returned_address,
phone_number,
location_type,
operational_status,
latitude,
longitude,
location_website,
location_overall_rating,
location_number_of_reviews,
monday_operating_hours,
tuesday_operating_hours,
wednesday_operating_hours,
thursday_operating_hours,
friday_operating_hours,
saturday_operating_hours,
sunday_operating_hours,
json_place_details_full_return,
-- measures
@@ -84,9 +471,10 @@ final__stg_exchange__company_addresses AS (
updated_datetime,
-- metadata
_function_ran_datetime,
_dbt_ran_datetime
FROM company_addressess_join_companies_countries_states_districts
)
FROM masked_data_transformation
)
SELECT * FROM final__stg_exchange__company_addresses
models/staging/exchange/stg_exchange__transaction_orders.sql
+4
View File
@@ -1,3 +1,7 @@
-- change name for final__stg_exchange__transaction_costs
-- @todo
-- @yam
-- IMPORTS
WITH transactions AS (
SELECT * FROM {{ ref('base_exchange__transactions') }}