Multiple Sources

Scenario

This example illustrates an example where multiple sources are used to populate a single hub. The three sources originate from a single table in the raw zone but use three different source table columns as representative business keys. The generate command creates three separate prep objects/dynamic tables which are used in the UNION when loading the final table.

Information Schema

All objects are based on the following information schema

Information Schema

TABLE_CATALOG,TABLE_SCHEMA,TABLE_NAME,COLUMN_NAME
YAML_TEMPLATES,SOURCE_DATA,MATERIAL,CDC_FLAG
YAML_TEMPLATES,SOURCE_DATA,MATERIAL,MATERIAL_PK
YAML_TEMPLATES,SOURCE_DATA,MATERIAL,MATERIAL_GLOBAL_ID
YAML_TEMPLATES,SOURCE_DATA,MATERIAL,CDC_TIMESTAMP
YAML_TEMPLATES,SOURCE_DATA,MATERIAL,MATERIAL_NAME
YAML_TEMPLATES,SOURCE_DATA,MATERIAL,MATERIAL_WEBSHOP_ID

Hub Definition

hub_material

entity_type: hub
name: HUB_MATERIAL

concatenate_business_keys: false
requires_source_business_key: false
enable_refresh: true

target_business_key_columns:
- MATERIAL_ID

entity_sources:

- urn:s2v:hub_source:src_1:
    entity_source: (SOURCE_DATA, MATERIAL)
    source_filter: ''
    source_system_configuration_urn: urn:s2v:source_setting:yaml_interface
    business_key_mapping:
    - MATERIAL_ID:
      - MATERIAL_PK
    source_business_key: ''

- urn:s2v:hub_source:src_2:
    entity_source: (SOURCE_DATA, MATERIAL)
    source_filter: ''
    source_system_configuration_urn: urn:s2v:source_setting:yaml_interface
    business_key_mapping:
    - MATERIAL_ID:
      - MATERIAL_GLOBAL_ID
    source_business_key: ''

- urn:s2v:hub_source:src_3:
    entity_source: (SOURCE_DATA, MATERIAL)
    source_filter: ''
    source_system_configuration_urn: urn:s2v:source_setting:yaml_interface
    business_key_mapping:
    - MATERIAL_ID:
      - MATERIAL_WEBSHOP_ID
    source_business_key: ''

Output - Prep Layer

------------------------------------------------------------------------------------
------ HUB_MATERIAL_[SOURCE_DATA,MATERIAL]_MAPPING_1
------------------------------------------------------------------------------------
CREATE OR REPLACE DYNAMIC TABLE "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}_PREP"."HUB_MATERIAL[SOURCE_DATA,MATERIAL]_MAPPING_1"
(
    "HKY_HUB_MATERIAL"
    ,"MATERIAL_ID"
    ,"LOAD_DATE" 
    ,"REC_SRC" 
)
    TARGET_LAG = DOWNSTREAM
    WAREHOUSE = &{TARGET_WAREHOUSE}
    REFRESH_MODE = INCREMENTAL
    INITIALIZE = ON_SCHEDULE
    WITH TAG ("&{TARGET_DATABASE}"."&{TARGET_SCHEMA}".S2V_RELEASE_ID='&{S2V_RELEASE_ID}')
AS
SELECT
    UPPER(SHA1(CONCAT(IFNULL(TRIM(REPLACE(CAST("MATERIAL_PK" as VARCHAR), '##', CONCAT('\\', '##'))), '-1'),'##'))) AS "HKY_HUB_MATERIAL"
    ,IFNULL(TRIM(REPLACE(CAST("MATERIAL_PK" as VARCHAR), '##', CONCAT('\\', '##'))), '-1') AS "MATERIAL_ID"
    ,MIN(TO_TIMESTAMP_LTZ("CDC_TIMESTAMP")) AS "LOAD_DATE" 
    ,'SOURCE_DATA.MATERIAL' AS "REC_SRC"
FROM "&{SOURCE_DATABASE}"."SOURCE_DATA"."MATERIAL"
GROUP BY ALL;

------------------------------------------------------------------------------------
------ HUB_MATERIAL_[SOURCE_DATA,MATERIAL]_MAPPING_2
------------------------------------------------------------------------------------
CREATE OR REPLACE DYNAMIC TABLE "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}_PREP"."HUB_MATERIAL[SOURCE_DATA,MATERIAL]_MAPPING_2"
(
    "HKY_HUB_MATERIAL"
    ,"MATERIAL_ID"
    ,"LOAD_DATE" 
    ,"REC_SRC" 
)
    TARGET_LAG = DOWNSTREAM
    WAREHOUSE = &{TARGET_WAREHOUSE}
    REFRESH_MODE = INCREMENTAL
    INITIALIZE = ON_SCHEDULE
    WITH TAG ("&{TARGET_DATABASE}"."&{TARGET_SCHEMA}".S2V_RELEASE_ID='&{S2V_RELEASE_ID}')
AS
SELECT
    UPPER(SHA1(CONCAT(IFNULL(TRIM(REPLACE(CAST("MATERIAL_GLOBAL_ID" as VARCHAR), '##', CONCAT('\\', '##'))), '-1'),'##'))) AS "HKY_HUB_MATERIAL"
    ,IFNULL(TRIM(REPLACE(CAST("MATERIAL_GLOBAL_ID" as VARCHAR), '##', CONCAT('\\', '##'))), '-1') AS "MATERIAL_ID"
    ,MIN(TO_TIMESTAMP_LTZ("CDC_TIMESTAMP")) AS "LOAD_DATE" 
    ,'SOURCE_DATA.MATERIAL' AS "REC_SRC"
FROM "&{SOURCE_DATABASE}"."SOURCE_DATA"."MATERIAL"
GROUP BY ALL;

------------------------------------------------------------------------------------
------ HUB_MATERIAL_[SOURCE_DATA,MATERIAL]_MAPPING_3
------------------------------------------------------------------------------------
CREATE OR REPLACE DYNAMIC TABLE "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}_PREP"."HUB_MATERIAL[SOURCE_DATA,MATERIAL]_MAPPING_3"
(
    "HKY_HUB_MATERIAL"
    ,"MATERIAL_ID"
    ,"LOAD_DATE" 
    ,"REC_SRC" 
)
    TARGET_LAG = DOWNSTREAM
    WAREHOUSE = &{TARGET_WAREHOUSE}
    REFRESH_MODE = INCREMENTAL
    INITIALIZE = ON_SCHEDULE
    WITH TAG ("&{TARGET_DATABASE}"."&{TARGET_SCHEMA}".S2V_RELEASE_ID='&{S2V_RELEASE_ID}')
AS
SELECT
    UPPER(SHA1(CONCAT(IFNULL(TRIM(REPLACE(CAST("MATERIAL_WEBSHOP_ID" as VARCHAR), '##', CONCAT('\\', '##'))), '-1'),'##'))) AS "HKY_HUB_MATERIAL"
    ,IFNULL(TRIM(REPLACE(CAST("MATERIAL_WEBSHOP_ID" as VARCHAR), '##', CONCAT('\\', '##'))), '-1') AS "MATERIAL_ID"
    ,MIN(TO_TIMESTAMP_LTZ("CDC_TIMESTAMP")) AS "LOAD_DATE" 
    ,'SOURCE_DATA.MATERIAL' AS "REC_SRC"
FROM "&{SOURCE_DATABASE}"."SOURCE_DATA"."MATERIAL"
GROUP BY ALL;

Output - Main Layer

------------------------------------------------------------------------------------
------ HUB_MATERIAL ---------------------------------------------------------------- 
------------------------------------------------------------------------------------
CREATE OR REPLACE DYNAMIC TABLE "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}"."HUB_MATERIAL"
(
    "HKY_HUB_MATERIAL"
    ,"MATERIAL_ID"
    ,"LOAD_DATE" 
    ,"REC_SRC" 
)
    TARGET_LAG = '&{TARGET_LAG}'
    WAREHOUSE = &{TARGET_WAREHOUSE}
    REFRESH_MODE = INCREMENTAL
    INITIALIZE = ON_SCHEDULE
    WITH TAG ("&{TARGET_DATABASE}"."&{TARGET_SCHEMA}".S2V_RELEASE_ID='&{S2V_RELEASE_ID}')
AS
SELECT
    "HKY_HUB_MATERIAL",
    "MATERIAL_ID",
    "LOAD_DATE",
    "REC_SRC"
FROM
    (
    SELECT
        *
    FROM "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}_PREP"."HUB_MATERIAL[SOURCE_DATA,MATERIAL]_MAPPING_1"
    UNION ALL
    SELECT
        *
    FROM "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}_PREP"."HUB_MATERIAL[SOURCE_DATA,MATERIAL]_MAPPING_2"
    UNION ALL
    SELECT
        *
    FROM "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}_PREP"."HUB_MATERIAL[SOURCE_DATA,MATERIAL]_MAPPING_3"
    UNION ALL
    SELECT
        OBJECT_H_KEY AS "HKY_HUB_MATERIAL",
        '-1' AS "MATERIAL_ID",
        "LOAD_DATE",
        RECORD_SOURCE AS "REC_SRC"
    FROM "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}_PREP"."GHOST_RECORD"
    )
QUALIFY ROW_NUMBER() OVER (PARTITION BY "HKY_HUB_MATERIAL" ORDER BY "LOAD_DATE" ASC ) = 1
;

Output - Stream

------------------------------------------------------------------------------------
------ STREAMS ON DT HUB_MATERIAL -------------------------------------------------- 
------------------------------------------------------------------------------------
CREATE OR REPLACE STREAM 
    "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}"."HUB_MATERIAL_STREAM" 
ON DYNAMIC TABLE "&{TARGET_DATABASE}"."&{TARGET_SCHEMA}"."HUB_MATERIAL" 
    SHOW_INITIAL_ROWS = TRUE;