Data Vault and Snowflake’s Metric Functions

A few years ago, we described what it takes to ensure that your data vault’s integrity remains in check. Data vault tables have certain data integrity requirements ensuring that:

• There is a unique representation of business keys in a hub table and a unique representation of a relationship in a link table
• True changes to a business entity’s state are recorded in a hub-satellite table and true changes to a relationship’s state are recorded in a link-satellite table.
• Hub satellite table records find an associated parent record in a hub table; link satellite table records find an associated parent record in a link table and hub participants in a link structure have associated hub records in those hub tables.

If a data platform supports enforcing primary key, foreign key and unique constraints then there is no need to check that these constraints exist in the data of those tables. Alas, Snowflake does not enforce these constraints in their proprietary storage and neither does the open-source Apache Iceberg standard. What data engineering teams must do is build these constraints into the data pipelines loading these data vault structures.

After a data vault table is loaded, we must ensure that the integrity of the data vault is still in check. Note that a hub table is a shared artefact in a data vault, it will get its business keys from several sources. Link tables can be multi-sourced too, but this depends on if you are really integrating multiple business processes into one link table. Raw vault satellite tables are always single source.

GitHub repo with this framework using Snowflake tables and streams,

https://github.com/PatrickCuba/the_data_must_flow/tree/master/data-auto-testing

What we are focused on for this article is Snowflake’s Data Metric functions as building for ensuring the integrity of your data vault.

First up…

How does Snowflake support Data Quality?

Conceptually Snowflake supports data quality through the amalgamation of several building blocks.

• Data metric function (DMF) — Snowflake provides out of the box system metrics (counts, min, max, average, standard deviation, accepted values) as well as the ability to create custom metrics. A DMF can be attached to a table or a view and explicitly define which columns the measure is applicable to.
• Expectations — A DMF attached to a table’s column will produce a measure, an expectation is what the outcome of that measure is expected to be. If we do not get the value we expect then we will need a method to alert data administrators of the failing measure. An expectation is defined on a measure.
• DMF schedule — executing a procedure to calculate the measure can be done on a schedule or alternatively we can trigger the execution of a DMF when the target table has new data. The latter is limited to physical tables and not views.
• Anomaly detection — paired with machine learning, Snowflake trains on the statistics you define in your DMFs and the expectation to spot anomalies in your data. Although we will not be using this in our data vault framework.

What you will notice right away is that these metrics are post data load operations, i.e. the problem has been introduced into the data vault and therefore you must work to remediate the problem to return the data vault to its correct state…

Now that we have the building blocks we need, let’s put them into practice.

Defining the DMFs for Data Vault

In the custom recon framework for data vault we did two things, count where we have expectations and measure to understand table shape. When understanding the shape of data, it is important to note that the shape of data depends on the aggregation of data and thus the output inherently becomes non-identifying. If you choose columns to aggregate on and your data pipeline receives a limited number of records, will it then be possible to use that output to re-identify the entity being aggregated? Use caution that you are not unintentionally creating a data compliance gap when introducing DMFs as DMFs store its output in Snowflake event tables.

Using DMFs to prove data vault table integrity

The custom recon framework is designed for:

• Ensuring that the hub, link and satellite tables do not have duplicate records by their respective table uniqueness constraints.
• Ensuring the data staged is the data loaded to the respective hub, link and satellite tables.
• Ensuring that satellite table parent hash keys are present in the parent tables they relate to. Link tables have their hub-hash keys loaded into hub tables and satellite tables have their hub or link hash keys loaded into hub and link tables.

Duplicate check

Hub tables form the integration between your enterprise software landscape (horizontal integration) and the business representation according to your business architecture (vertical integration). The uniqueness of each hub table will be measured using a custom DMF despite the existence of a system DMF called DUPLICATE_COUNT. The reason for this is because the system duplicate count has the following two limitations:

• System DMFs do not support the binary data type, and
• Duplicate count can only be called when processing on a single column

We have long established that hash keys and hashdiffs (record digests) are always deployed as a binary data type. By sticking to this standard, you promote good data architecture practice by enforcing in-database execution of table joins and other operations on your data vault model. Hash keys must never leave the data platform.

Define Duplicate Check DMF for Hubs

If your data vault does not use surrogate hash keys then you only need one DMF assigned to a hub table, but if you do use surrogate hash keys, then you will require two DMFs assigned to a hub table:

• Check the uniqueness of the surrogate hub-hash key.
• Check the uniqueness of the tenant-id + business key collision code + 1 (or more) business keys.

If the former returns zero but the latter checks returns with a value greater than zero then you have one more problem than just a non-unique hub table!

-- hub-hash key

CREATE DATA METRIC FUNCTION DV_DMF_HUB_SKEY_DUPE_err(
    arg_t TABLE(skey BINARY)
)
RETURNS NUMBER
COMMENT = 'DV2 HUB: Count of rows where the hub surrogate hash key is duplicated. Expectation: 0.'
AS $$
    SELECT COUNT(*) FROM (
        SELECT skey FROM arg_t
        GROUP BY skey HAVING COUNT(*) > 1
    )
$$;

-- one business key

CREATE DATA METRIC FUNCTION DV_DMF_HUB_1BKEY_DUPE_err(
    arg_t TABLE(tenant_id VARCHAR, bkeycolcode VARCHAR, bkey1 VARCHAR)
)
RETURNS NUMBER
COMMENT = 'DV2 HUB: Count of rows where (tenant_id, bkeycolcode, bkey1) composite is duplicated. Expectation: 0.'
AS $$
    SELECT COUNT(*) FROM (
        SELECT tenant_id, bkeycolcode, bkey1 FROM arg_t
        GROUP BY tenant_id, bkeycolcode, bkey1 HAVING COUNT(*) > 1
    )
$$;

Each DMF has a unique signature and if you have hub tables with composite business keys then you will need additional DMFs to support it.

ALTER TABLE HUB_ACCOUNT
    ADD DATA METRIC FUNCTION DV_DMF_HUB_SKEY_DUPE_err
        ON (DV_HASHKEY_HUB_ACCOUNT)
        EXPECTATION hub_account_skey_no_dupes (VALUE = 0);

ALTER TABLE HUB_ACCOUNT
    ADD DATA METRIC FUNCTION DV_DMF_HUB_1BKEY_DUPE_err
        ON (DV_HASHKEY_HUB_ACCOUNT)
        EXPECTATION hub_some_table_1bkey_no_dupes (VALUE = 0);

Hub tables will be populated by multiple data pipelines multiple times a day and multiple times within the same data pipeline as well. Do you repeatedly run duplicate check DMFs every time the hub table is updated, or do you schedule the duplicate check DMF to run once a day? Perhaps introduce a warranty period for new data pipelines that after a few days you switch from trigger on change for the hub table to a schedule you run once a day. Remember, DMF schedules are defined on the table and not by data pipeline.

Define Duplicate Check DMF for Links

A link table in a hash-key data vault will require two DMFs assigned to it, and their categories are:

• Check the uniqueness of the surrogate link-hash key. This signature is identical to how we check hub-hash key uniqueness in a hub table and therefore you should reuse the DMF “DV_DMF_HUB_SKEY_DUPE_err”.
• Check the uniqueness of two or more hub-hash-keys that participate in that relationship

The link hash key is calculated by combining the columns needed to produce the hub hash keys used to represent the relationship; if you find that either check returns a value greater than zero then (just like the hub duplicate check we described before) you have one more problem to solve than just a non-unique link table. Note that these patterns do not apply to non-historized link tables!

-- two hub hash keys

CREATE DATA METRIC FUNCTION DV_DMF_LNK_2HKEY_DUPE_err(
    arg_t TABLE(hkey1 BINARY, hkey2 BINARY)
)
RETURNS NUMBER
COMMENT = 'DV2 LNK: Count of rows where (hkey1, hkey2) hub FK combination is duplicated. Expectation: 0.'
AS $$
    SELECT COUNT(*) FROM (
        SELECT hkey1, hkey2 FROM arg_t
        GROUP BY hkey1, hkey2 HAVING COUNT(*) > 1
    )
$$;

You will undoubtedly have link tables with more than just two hub table participants. For each configuration your data vault supports, you will need to create additional custom DMFs to measure them! Below is an example of a business vault link table with five hub table participants!

ALTER TABLE LNK_BV_CARD_ACCOUNT_ASSIGNMENT
    ADD DATA METRIC FUNCTION DV_DMF_LNK_5HKEY_DUPE_err
        ON (DV_HASHKEY_HUB_ACCOUNT_ACCOUNT_ID,
            DV_HASHKEY_HUB_ACCOUNT_BC_CONSOL_PRIMARY_ACCT,
            DV_HASHKEY_HUB_ACCOUNT_BC_TRANSFER_ACCOUNT_NO,
            DV_HASHKEY_HUB_ACCOUNT_CARD_ID,
            DV_HASHKEY_HUB_ACCOUNT_CONTROL_CARD_ID)
        EXPECTATION lnk_bvcaa_5hkey_no_dupes (VALUE = 0);

Just like running DMFs for hub duplicate checking, consider a warranty period for how frequently you will run this check.

Define the DMF for Satellites

Underneath it all, Snowflake is running a query using Snowflake’s vectorized query execution engine to return duplicate counts. While hub and link tables tend to be small, satellite tables can grow to billions if not trillions of records. We need to ensure that the duplicate check only runs on the data that was just added. The column that naturally clusters in a file-based table is the load date timestamp. Designing the custom DMF query to prune by this column ensures we achieve maximum efficiency to run said duplicate check. We achieve this by inserting a max(load-timestamp) clause, this query will cost you nothing as it is a metadata query.

At a minimum, you will need three DMFs to check for duplicate satellite table records:

• Regular satellite tables — this will also support effectivity, record tracking and status tracking satellite tables.
• Satellite tables with one or more dependent-child keys. If you have modelled satellite tables with two dependent-child keys then this will have to be supported by an additional DMF.
• Multi-active satellite tables.

-- Regular satellite

CREATE DATA METRIC FUNCTION DV_DMF_SAT_SKEY_DUPE_err(
    arg_t TABLE(skey BINARY, load_ts TIMESTAMP_NTZ, hashdiff BINARY))
RETURNS NUMBER
COMMENT = 'DV2 SAT: Count of duplicate (skey, load_ts, hashdiff) in the latest loaded batch only. Expectation: 0.'
AS $$
    SELECT COUNT(*) FROM (
        SELECT skey, load_ts, hashdiff FROM arg_t
        WHERE load_ts = (SELECT MAX(load_ts) FROM arg_t)
        GROUP BY skey, load_ts, hashdiff HAVING COUNT(*) > 1
    )
$$;

-- Satellite table with 1 dependent-child key

CREATE DATA METRIC FUNCTION DV_DMF_DSAT_1SKEY_DUPE_err (
    arg_t TABLE(skey BINARY, dep_key1 VARCHAR, load_ts TIMESTAMP_NTZ, hashdiff BINARY))
RETURNS NUMBER
COMMENT = 'DV2 DSAT: Count of duplicate (skey, dep_key1, load_ts) in the latest loaded batch only. Expectation: 0.’
AS $$
    SELECT COUNT(*) FROM (
        SELECT skey, dep_key1, load_ts, hashdiff FROM arg_t
        WHERE load_ts = (SELECT MAX(load_ts) FROM arg_t)
        GROUP BY skey, dep_key1, load_ts, hashdiff HAVING COUNT(*) > 1
    )
$$;

-- Multi-active satellite table

CREATE DATA METRIC FUNCTION DV_DMF_MSAT_SKEY_DUPE_err(
    arg_t TABLE(skey BINARY, sequence NUMBER, load_ts TIMESTAMP_NTZ, hashdiff BINARY))
RETURNS NUMBER
COMMENT = 'DV2 MSAT: Count of duplicate (skey, sequence, load_ts) in the latest loaded batch only. Expectation: 0.'
AS $$
    SELECT COUNT(*) FROM (
        SELECT skey, sequence, load_ts, hashdiff FROM arg_t
        WHERE load_ts = (SELECT MAX(load_ts) FROM arg_t)
        GROUP BY skey, sequence, load_ts, hashdiff HAVING COUNT(*) > 1
    )
$$;

Unlike checking for duplicate records in hub or link tables, duplicate checks for satellite tables are intentionally based on the latest slice of satellite data. Therefore, if you run more than one data load to a satellite table and run a duplicate check just once, you will only be testing for duplicates on the latest load timestamp slice. Once again, consider if you will use a warranty period for running duplicate checks on satellite tables, the difference here is that raw vault satellite tables are always single source.

ALTER TABLE SAT_BV_CREDITSCORE
    ADD DATA METRIC FUNCTION DV_DMF_SAT_SKEY_DUPE_err
        ON (DV_HASHKEY_HUB_PARTY, DV_LOAD_TIMESTAMP, DV_TENANT_ID, DV_HASHDIFF)
        EXPECTATION sat_bvc_skey_no_dupes (VALUE = 0);
ALTER TABLE SAT_BV_CREDITSCORE
    SET DATA_METRIC_SCHEDULE = 'TRIGGER_ON_CHANGES';

Referential integrity check

A data vault model will contain many tables to join on; a flawlessly modelled data vault will utilise zero-keys to cater for the different cardinalities of relationships between tables. By design, the data vault is eventually consistent because there are no dependencies between data pipelines and no dependencies even within a data pipeline as it will be loading related hub, link and satellite tables. This means that this category of DMFs cannot be executed as soon as their respective data loads complete because if they do they may be reporting false positives as the related tables are being populated and the DMF is executing on a read-committed transaction. For the DMFs you see below, we recommend running them on a schedule.

These are the referential integrity rules we must check for:

• A link table will have one or more hub tables it is related to and two or more hub hash keys we must run an orphan check on. For each combination of hub-hash keys to check we will use the same DMF but applied to different column comparison configurations.
• Satellite tables are far simpler to check, either the table is a hub-satellite, and we check for missing surrogate hash keys between the hub and satellite, or the table is a link-satellite, and we check for missing surrogate hash keys between the link and satellite.

-- Link table orphans

CREATE DATA METRIC FUNCTION DV_DMF_LNK_SKEY_ORPH_err(
    arg_lnk TABLE(fk_col BINARY, 
    arg_hub TABLE(pk_col BINARY))
RETURNS NUMBER
COMMENT = 'DV2 LNK ORPHAN: Count of latest-batch LNK FK keys not found in parent HUB. Expectation: 0.'
AS $$
    SELECT COUNT(*) FROM arg_lnk l
    WHERE NOT EXISTS (
          SELECT 1 FROM arg_hub h WHERE h.pk_col = l.fk_col)
$$;

-- Satellite table orphans

CREATE DATA METRIC FUNCTION DV_DMF_SAT_SKEY_ORPH_err(
    arg_sat TABLE(fk_col BINARY, rec_source VARCHAR),
    arg_parent TABLE(pk_col BINARY))
RETURNS NUMBER
COMMENT =  'DV2 SAT ORPHAN: Count of latest-batch SAT FK keys not found in parent HUB or LNK, excluding GHOST records. Expectation: 0.'
AS $$
SELECT COUNT(*) FROM arg_sat s
    WHERE s.rec_source <> 'GHOST'
      AND NOT EXISTS (
          SELECT 1 FROM arg_parent p WHERE p.pk_col = s.fk_col)
$$;

Custom DMFs have a two table limit you can refer to; thus, each referential integrity will require an instance of the DMF applied, the name of the expectation is what will make the DMF to table relationship unique. It is essentially the name of the instance.

ALTER TABLE LNK_BV_CARD_ACCOUNT_ASSIGNMENT
    ADD DATA METRIC FUNCTION DV_DMF_LNK_SKEY_ORPH_ERR
    ON (DV_HASHKEY_HUB_ACCOUNT_CARD_ID,
        TABLE(HUB_ACCOUNT(DV_HASHKEY_HUB_ACCOUNT)))
    EXPECTATION lnk_bvcaa_card_id_orph (VALUE = 0);

ALTER TABLE LNK_BV_CARD_ACCOUNT_ASSIGNMENT
    ADD DATA METRIC FUNCTION DV_DMF_LNK_SKEY_ORPH_ERR
    ON (DV_HASHKEY_HUB_ACCOUNT_BC_TRANSFER_ACCOUNT_NO,
        TABLE(HUB_ACCOUNT(DV_HASHKEY_HUB_ACCOUNT)))
    EXPECTATION lnk_bvcaa_bc_transfer_orph (VALUE = 0);

ALTER TABLE LNK_BV_CARD_ACCOUNT_ASSIGNMENT
    ADD DATA METRIC FUNCTION DV_DMF_LNK_SKEY_ORPH_ERR
    ON (DV_HASHKEY_HUB_ACCOUNT_BC_CONSOL_PRIMARY_ACCT,
        TABLE(HUB_ACCOUNT(DV_HASHKEY_HUB_ACCOUNT)))
    EXPECTATION lnk_bvcaa_bc_consol_orph (VALUE = 0);

Reconciliation check

After each data pipeline loads their respective hub, link and satellite tables, we must provide evidence that all staged data exists in the target tables. A staged universe will load:

• One or more hub tables. All business keys and their associated surrogate hash keys, tenant ids and collision codes in staging are either already present in the designated target hub table or loaded as new business keys with new load-timestamps.
• Loading to link tables will have the same semantics as loading to hub tables, either the staged relationship already exists in the target link table or new relationships will be loaded with a new load timestamp.
• Loading hub and link satellite tables are a little different, the staged attributes describing a business object or relationship is compared to the target satellite table’s current active records for that business object or relationship. If the descriptive attributes are the same, the staged record is not loaded, but if they differ then they will be loaded.

Thus, the reconciliation evidence between what was staged and what exists in the data vault must show that the staging content can be discarded without any fear of data loss. Given that we advocate that the data vault can be continuously loaded, landed content can be truncated as we can be assured that the data has been loaded. To gather this evidence, reconciliation must therefore be executed as soon as the independent data pipeline to a hub, link or satellite table has completed and as quickly as possible. As a Snowflake DMF this means that the schedule a DMF is applied to the target table must execute when the target table changes. And herein lies the problem with this architectural design;

• What if the new staged content does not introduce new data into a target data vault table structure?
• How will we know that the data pipeline was executed successfully if no DMF check is triggered?

Scheduling the execution of a DMF could likely miss multiple data pipeline executions and attempting to schedule DMF execution for every 5 minutes to circumvent this limitation is inaccurate and clumsy. The only reasonable idempotent method to execute reconciliation between staged content and target data vault tables is to stick to the custom framework we have already established a few years ago.

While there we can gather the shape of data metrics present in staging and utilize Snowflake streams on top of the target data vault tables to gather evidence of what records are new.

Although it is possible to define the DMFs for reconciliation and execute them using the “SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS” system call, using this function will fire every DMF defined on top of a target data vault table. For a hub table this can be many and for a referential integrity check you might produce false positives.

If your data vault framework is not continuously loading to the target tables, here are the system DMFs you can attach to your hub, link and satellite tables.

-- on satellite table: staged content exist, for sats with dep-keys add the dep-key

CREATE DATA METRIC FUNCTION DV_DMF_SAT_HDIF_SGTG_err(
   arg_sat TABLE(skey2 BINARY, hashdiff2 BINARY,
                  applied_ts TIMESTAMP_NTZ, load_ts TIMESTAMP_NTZ),
arg_stg TABLE(skey1 BINARY, hashdiff1 BINARY)

)
RETURNS NUMBER
COMMENT = 'DV2 SAT RECON: Count of staged (skey+hashdiff) not present in current SAT record (single-source RV). Expectation: 0.'
AS $$
   SELECT COUNT(*) FROM arg_stg s
    WHERE NOT EXISTS (
        SELECT 1 FROM (
            SELECT skey, tenant_id, hashdiff FROM arg_sat
            QUALIFY RANK() OVER (
                PARTITION BY skey
                ORDER BY applied_ts DESC, load_ts DESC
            ) = 1
        ) curr
        WHERE curr.skey   = s.skey
          AND curr.hashdiff  = s.hashdiff
    )
$$;

-- DMFs for staging

ALTER TABLE SAT_RV_HUB_SAPBW_COMM_CUSTOMER
    ADD DATA METRIC FUNCTION DV_DMF_SAT_HDIF_SGTG_err
    ON (DV_HASHKEY_HUB_PARTY,                               -- → skey2
        DV_HASHDIFF,        -- → hashdiff2
        DV_APPLIED_TIMESTAMP, 
        DV_LOAD_TIMESTAMP                         
        TABLE(STG_ SAPBW_COMM_CUSTOMER(
            DV_HASHKEY_HUB_PARTY,                           -- → skey1
            DV_HASHDIFF SAT_RV_HUB_SAPBW_COMM_CUSTOMER)))   -- → load_ts
    EXPECTATION sat_comm_cust_hdif_from_stg (VALUE = 0);

*Expand to support daily metrics by adding DMFs to staging and to support satellite tables with dependent-child keys.

Visibility options

Since Snowflake DMFs are a native utility there are built-in features we can use to surface the data quality and integrity issues. We will recommend that depending on the metric, we should consider two types reporting:

· Alerts when something breaks

· Reports to show everything is running smoothly

Event Tables

To retrieve data metric functions statistics and violations you can query the following Snowflake supplied event table and views under your Snowflake account’s snowflake.local schema:

· Event table: DATA_QUALITY_MONITORING_RESULTS_RAW

· Views: DATA_QUALITY_MONITORING_EXPECTATION_STATUS

· Views: DATA_QUALITY_MONITORING_RESULTS

Webhooks

If your client supports webhooks then it is likely Snowflake can send notifications to it.

In Slack:

1. Create a Slack App by visiting api.slack.com/apps, create a new App from scratch.
2. Enable incoming Webhooks (Features, incoming Webhooks, toggle on, add a new Webhook to Workspace and select the channel).
3. Copy the webhook URL from Slack — the path’s suffix goes into the Snowflake secret.

In Snowflake:

1. Create secret
2. Create notification integration
3. Create stored procedure
4. Create alert

-- Secret
CREATE SECRET SLACK_DV_DQ_WEBHOOK_SECRET
    TYPE = GENERIC_STRING
    SECRET_STRING = ‘XXX';

-- Notification Integration
CREATE NOTIFICATION INTEGRATION SLACK_DV_DQ_ALERTS
    ENABLED = TRUE
    TYPE = WEBHOOK
    WEBHOOK_URL = 'https://hooks.slack.com/services/SNOWFLAKE_WEBHOOK_SECRET'
    WEBHOOK_SECRET = SLACK_DV_DQ_WEBHOOK_SECRET
    WEBHOOK_BODY_TEMPLATE = '{"text": "SNOWFLAKE_WEBHOOK_MESSAGE"}'
    WEBHOOK_HEADERS = ('Content-Type'='application/json');

-- When something breaks
CREATE PROCEDURE SP_DQ_VIOLATION_ALERT()
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS $$
DECLARE
    msg       VARCHAR DEFAULT '';
    fail_cnt  INTEGER DEFAULT 0;
    c1 CURSOR FOR
        SELECT
            TABLE_DATABASE || '.' || TABLE_SCHEMA || '.' || TABLE_NAME AS table_ref,
            METRIC_NAME,
            EXPECTATION_NAME,
            VALUE
        FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_EXPECTATION_STATUS
        WHERE EXPECTATION_VIOLATED = TRUE
          AND METRIC_NAME LIKE '%_ERR'
          AND SCHEDULED_TIME >= DATEADD(minute, -10, CURRENT_TIMESTAMP())
        ORDER BY TABLE_NAME, METRIC_NAME;
BEGIN
    FOR rec IN c1 DO
        fail_cnt := fail_cnt + 1;
        msg := msg
            || chr(10) || ':x:  *' || rec.table_ref || '*'
            || chr(10) || '    Metric: '      || rec.METRIC_NAME
            || chr(10) || '    Expectation: ' || rec.EXPECTATION_NAME
            || chr(10) || '    Value: *'      || rec.VALUE || '*'
            || chr(10);
    END FOR;

    IF (fail_cnt > 0) THEN
        msg := ':rotating_light: *DV DQ VIOLATION — ' || fail_cnt || ' expectation(s) failed*'
            || chr(10) || msg;
        CALL SYSTEM$SEND_SNOWFLAKE_NOTIFICATION(
            SNOWFLAKE.NOTIFICATION.TEXT_PLAIN(
                SNOWFLAKE.NOTIFICATION.SANITIZE_WEBHOOK_CONTENT(
                    REPLACE(:msg, chr(10), chr(92)||'n')
                )
            ),
            SNOWFLAKE.NOTIFICATION.INTEGRATION('SLACK_DV_DQ_ALERTS')
        );
    END IF;

    RETURN msg;
END;
$$;

-- Daily report
CREATE PROCEDURE SP_DQ_DAILY_REPORT()
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS $$
DECLARE
    msg       VARCHAR DEFAULT '';
    pass_cnt  INTEGER DEFAULT 0;
    fail_cnt  INTEGER DEFAULT 0;
    failures  VARCHAR DEFAULT '';
    c_fail CURSOR FOR
        SELECT
           TABLE_DATABASE || '.' || TABLE_SCHEMA || '.' || TABLE_NAME AS table_ref,
            METRIC_NAME,
            EXPECTATION_NAME,
            VALUE
       FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_EXPECTATION_STATUS
        WHERE METRIC_NAME LIKE '%_ERR'
          AND SCHEDULED_TIME >= DATEADD(minute, -24, CURRENT_TIMESTAMP())
          AND EXPECTATION_VIOLATED = TRUE
        ORDER BY TABLE_NAME, METRIC_NAME;
BEGIN
    FOR rec IN c_fail DO
        fail_cnt := fail_cnt + 1;
        failures := failures
            || chr(10) || '   :x:  *' || rec.table_ref || '*'
            || ' / ' || rec.METRIC_NAME
            || ' — Value: *' || rec.VALUE || '*';
    END FOR;

    SELECT COUNT(*) INTO :pass_cnt
   FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_EXPECTATION_STATUS
        WHERE METRIC_NAME LIKE '%_ERR'
          AND SCHEDULED_TIME >= DATEADD(minute, -24, CURRENT_TIMESTAMP())
          AND EXPECTATION_VIOLATED = FALSE;

    msg := ':bar_chart: *DV DQ Daily Report — ' || TO_CHAR(CURRENT_DATE(), 'YYYY-MM-DD') || '*'
        || chr(10) || ':white_check_mark:  Passing: *' || pass_cnt || '*'
        || chr(10) || ':x:  Failing: *' || fail_cnt || '*';

    IF (fail_cnt > 0) THEN
        msg := msg || chr(10) || chr(10) || '*Failures:*' || failures;
    ELSE
        msg := msg || chr(10) || chr(10) || ':tada:  All checks passed today.';
    END IF;

    CALL SYSTEM$SEND_SNOWFLAKE_NOTIFICATION(
        SNOWFLAKE.NOTIFICATION.TEXT_PLAIN(
            SNOWFLAKE.NOTIFICATION.SANITIZE_WEBHOOK_CONTENT(
                REPLACE(:msg, chr(10), chr(92)||'n')
            )
        ),
        SNOWFLAKE.NOTIFICATION.INTEGRATION('SLACK_DV_DQ_ALERTS')
    );

    RETURN msg;
END;
$$;

-- Alert
CREATE ALERT ALERT_DQ_VIOLATION
    WAREHOUSE = ADMIN_XSMALL
    SCHEDULE = '5 MINUTES'
    IF (EXISTS (
        SELECT 1
        FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_EXPECTATION_STATUS
        WHERE EXPECTATION_VIOLATED = TRUE
          AND METRIC_NAME LIKE '%_ERR'
          AND SCHEDULED_TIME >= DATEADD(minute, -10, CURRENT_TIMESTAMP())
    ))
    THEN CALL SP_DQ_VIOLATION_ALERT();

ALTER ALERT ALERT_DQ_VIOLATION RESUME;

-- Report
CREATE ALERT ALERT_DQ_DAILY_REPORT
    WAREHOUSE = ADMIN_XSMALL
    SCHEDULE = 'USING CRON 0 6 * * * UTC'
    IF (EXISTS (SELECT 1))
    THEN CALL SP_DQ_DAILY_REPORT();

ALTER ALERT ALERT_DQ_DAILY_REPORT RESUME;

Example alerts

Email

If you prefer sending emailed alerts and reports then the process is far shorter but just as simple. Set up an email integration and use the components listed above to send those notifications.

CREATE NOTIFICATION INTEGRATION EMAIL_DV_DQ_ALERTS
    TYPE = EMAIL
    ENABLED = TRUE
    ALLOWED_RECIPIENTS = ('you@yourcompany.com', 'team@yourcompany.com');

Snowsight

Snowflake has introduced Data Quality tabs you can access per table or view; you can even use Cortex to analyse and recommend data quality checks.

Wrap up

Snowflake DMF and data quality framework is a great facility for measuring and validating your data that we have shown can be utilised on a data vault. We have shown that DMFs can be used successfully to accurately track two of the three tests of the data vault test framework and if your data vault loading patterns then the full range can be covered.

To achieve minimal data quality checks needed to guarantee data vault integrity you could also consider applying only the following DMFs.

· Recon between staged descriptive content to target satellite tables

· Referential integrity checks between satellite and parent hub or link tables and between link and parent hub tables.

This at least proves that the data has loaded and that you can equi-join between your data vault tables and everything in staging was loaded. The net effect is a reduction in orchestrated data pipelines, and it will also support accurate construction of your PITs, Bridges and SNOPITs for your information marts as well.

The views expressed in this article are that of my own, you should test implementation performance before committing to this implementation. The author provides no guarantees in this regard.