diff --git a/README.md b/README.md index 37b27fe..3bead35 100644 --- a/README.md +++ b/README.md @@ -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.
-Short naming converntion lists that need to follow +Short naming convention lists that need to follow | model_type Shortcut | Full name | | ------------------- | ------ | diff --git a/dbt_project.yml b/dbt_project.yml index 4582ccc..fa4cd16 100644 --- a/dbt_project.yml +++ b/dbt_project.yml @@ -104,4 +104,6 @@ tests: +severity: warn - +vars: + mask_data_pattern: "('*MASK*', '****')" + mask_data_pattern_for_udf: "[*MASK*, ****]" diff --git a/macros/_macro__docs.md b/macros/_macro__docs.md new file mode 100644 index 0000000..098543a --- /dev/null +++ b/macros/_macro__docs.md @@ -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: ______________________________________________________ +``` +
+ +## 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: `` + - 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 Step 1: Cloud function trigger URL. For example: `https://us-central1-data-management-123456.cloudfunctions.net/address_to_geo_coordinate` + +
+ +### 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: + /: + post: + summary: Echo the input. + operationId: echo + x-google-backend: + address: + protocol: h2 + responses: + '200': + description: + schema: + type: string + ``` + +3. Under paths, replace `` 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 Step 2: Path Suffix. For example, `func-address-to-geo` +5. Replace `` with the value previously recorded in the template field Step 1: Cloud Function Trigger URL ***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 Step 3: Configuration File Name. 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: `` + - API ID: `` + - 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: `` + - Select a Service Account: App Engine default service account + + - **Gateway details** + - Display Name: `` + - 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 Step 3: Managed Service Identifier. 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 Step 3: Gateway Base URL. 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 + api_provider = google_api_gateway + google_audience = '' + api_allowed_prefixes = ('') + enabled = true; +``` + +4. Replace `` with `` and record the newly defined name into the template field Step 4: API Integration Name. For example, `gcp_api_integration_address_to_coordinates` +5. Replace `` with value previously recorded from template field Step 3: Managed Service Identifier +6. Execute the statement. +7. Run the following sql statement: + + ```sql + describe integration ; + ``` + +8. Record `API_GCP_SERVICE_ACCOUNT` value obtained from the execution into the template field Step 4: API_GCP_SERVICE_ACCOUNT. 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 () + returns variant + api_integration = + as ''; + ``` + +2. Replace `` with a unique function name of your choice. For example, `echo_address_to_coordinates` +3. Replace `` based on your external function configuration. For example, `id VARCHAR, full_address VARCHAR` +4. Replace `` with the value previously recorded from template field Step 4: API Integration Name +5. Replace `` with the value previously recorded from template field Step 3: Gateway Base URL and Step 2: Path Suffix. 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. + +
+ +## Security Policy + +
+ +## 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; +``` + +
+ +## 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`
+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`
+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`
+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`
+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`
+This CTE combines the results from all the batches into a single table. + +`operation_hours_json`
+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`
+This CTE further processes the extracted operating hour ranges and removes the day prefixes. + +`operation_hour_to_24_hour_transform`
+This CTE converts the operating hour ranges to a 24-hour format. + +`special_single_operation_hour_to_24_hour_transform`
+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`
+This CTE handles cases where there are two operating hours within a day and converts them to a 24-hour format. + +`concatenate_operation_hour`
+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`
+This CTE further extracts potentially valuable information returned by the API and flatten them. + +`masked_data_transformation`
+This CTE applies transformations to the selected metrics and dimensions. It replaces sensitive information with dummy values if the address is masked. + + +{% enddocs %} diff --git a/macros/_macro__models.yml b/macros/_macro__models.yml new file mode 100644 index 0000000..75dde26 --- /dev/null +++ b/macros/_macro__models.yml @@ -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. \ No newline at end of file diff --git a/macros/address_to_coordinates.sql b/macros/address_to_coordinates.sql new file mode 100644 index 0000000..708cea4 --- /dev/null +++ b/macros/address_to_coordinates.sql @@ -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 -%} \ No newline at end of file diff --git a/models/marts/reporting/rep_exchange__daily_orders.sql b/models/marts/reporting/rep_exchange__daily_orders.sql index b1d41d7..e2200bb 100644 --- a/models/marts/reporting/rep_exchange__daily_orders.sql +++ b/models/marts/reporting/rep_exchange__daily_orders.sql @@ -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, diff --git a/models/marts/warehouse/_warehouse_models.yml b/models/marts/warehouse/_warehouse_models.yml index 69662cd..fdb915d 100644 --- a/models/marts/warehouse/_warehouse_models.yml +++ b/models/marts/warehouse/_warehouse_models.yml @@ -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 diff --git a/models/marts/warehouse/dim_exchange__companies.sql b/models/marts/warehouse/dim_exchange__companies.sql index 1875a19..d27d5b9 100644 --- a/models/marts/warehouse/dim_exchange__companies.sql +++ b/models/marts/warehouse/dim_exchange__companies.sql @@ -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, diff --git a/models/staging/exchange/_exchange__models.yml b/models/staging/exchange/_exchange__models.yml index 4477bbb..03ecce8 100644 --- a/models/staging/exchange/_exchange__models.yml +++ b/models/staging/exchange/_exchange__models.yml @@ -23,4 +23,6 @@ models: description: Bridge between `companies` and `users` - name: stg_exchange__company_addresses - description: Company addresses that input when register \ No newline at end of file + description: Company addresses that input when register + config: + tags: ['incremental'] \ No newline at end of file diff --git a/models/staging/exchange/stg_exchange__company_addresses.sql b/models/staging/exchange/stg_exchange__company_addresses.sql index fe4ba5b..982c31c 100644 --- a/models/staging/exchange/stg_exchange__company_addresses.sql +++ b/models/staging/exchange/stg_exchange__company_addresses.sql @@ -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 \ No newline at end of file diff --git a/models/staging/exchange/stg_exchange__transaction_orders.sql b/models/staging/exchange/stg_exchange__transaction_orders.sql index a05aa7a..7f8a529 100644 --- a/models/staging/exchange/stg_exchange__transaction_orders.sql +++ b/models/staging/exchange/stg_exchange__transaction_orders.sql @@ -1,3 +1,7 @@ +-- change name for final__stg_exchange__transaction_costs +-- @todo +-- @yam + -- IMPORTS WITH transactions AS ( SELECT * FROM {{ ref('base_exchange__transactions') }}