Part II or Stale data detection with dbt and Redshift metadata

In an ideal world, your SaaS provider, e.g. Xero allows you to export your data or has some systems in place that push data to your preferred storage, e.g. to your data warehouse. In the real world, the shared data might be outdated which could impact the quality of data analysis, and hence, the quality of your decisions. And this is where data observability comes into the picture.

Data observability is an organization’s ability to fully understand the health of the data in their systems to ensure little or no anomalous data is fresh and mapped to the source data.

Source: https://www.moderndatastack.xyz/category/data-quality-monitoring

We would just take the ‘update’ column and add freshness tests with dbt. Great, we are sorted, right? Well, no. The challenge with off-the-shelf SaaS applications is that it is not always possible to add a column to the raw data to indicate when the data was pushed to your systems.

To overcome this challenge, I looked into the metadata tables as they often have the “creation / modified date”. In Part I or “State data detection with dbt and BigQuery dataset metadata”. I already checked BigQuery and dbt. What makes Redshift different from BigQuery is that pg_class_info doesn’t contain “last modified time” of the data warehouse tables. As for the system catalog tables, they are maintained only for 2–5 days. AWS recommends “to periodically copy it to other tables or unload it to Amazon S3”. Then, you could import these files from s3 into a permanent history table in your data warehouse and use it to identify the date when the imported table was modified last time. If you are using AWS services, check out “Export RedShift System Tables And Views To S3” on how to export Redshift System tables to S3 using the stored procedures. The article also contains other links on how to complete the task with AWS Lambda and AWS Glue.

In my case, as dbt is a preferred ELT tool, I decided to use it to extract the previous day's records from the query and insert tables and append them to the existing history tables. If you don’t know what dbt is, dbt or data build tool is an Extract Load Transform (ELT) tool developed by a consulting company Fishtown Analytics, currently named dbt Labs. The company has two product offerings: dbt Core, an open-source command-line interface (CLI) and dbt Cloud. For simplicity, I would be using dbt Cloud.

Getting ready

1. There are already plenty of great tutorials on the Internet on how to set up dbt with Redshift:
   — Quickstart for dbt Cloud and Redshift — https://docs.getdbt.com/docs/quickstarts/dbt-cloud/redshift
   — dbt CLI and Amazon Redshift workshop— https://catalog.workshops.aws/dbt-cli-and-amazon-redshift/en-US/introduction

How to do it…

1. Create a new file, e.g. stl_insert_history.sql under models/ and paste the following code. In this code we extract all insert/delete/create records populated in the stl_insert table as of yesterday and append them to the table stl_insert_history in the schema system_tables. Incremental models are one of four materialization options in dbt where “first time a model is run, the table is built by transforming all rows of source data. On subsequent runs, dbt transforms only the rows in your source data that you tell dbt to filter for, inserting them into the target table which is the table that has already been built. (dbt)” During the first run, the table will be built with all rows in the table, however, as we don’t specify a key column, dbt will simpler append new data to the existing table.

{{ config(materialized = 'incremental', alias = 'stl_insert_history', schema = 'system_tables', tags = ["daily"]) }}

select *, 
CAST('{{run_started_at.strftime ("%Y-%m-%d %H:%M:%S") }}' AS TIMESTAMP) as dbt_at
from stl_insert
where starttime::date = current_date-1

2. Create a new file, e.g. stl_query_history.sql under models/ and paste the following code. Like in the case with stl_insert, we extract all queries that were run yesterday and append them to the table stl_query_history in the schema system_tables — you could extract the entire table but the size of your table will grow exponentially with the increase of the usage of the data warehouse.

{{ config(materialized = 'incremental',
 alias = 'stl_query_history', schema = 'system_tables', tags = ["daily"]) }}

select *, 
CAST('{{run_started_at.strftime ("%Y-%m-%d %H:%M:%S") }}' AS TIMESTAMP) as dbt_at
from stl_query
where starttime::date = current_date-1

If you run this query in your data warehouse, you would get the list of tables in your data warehouse and the last update date — provided that your tables got updated/created after you started capturing the history.

WITH max_query_time
     AS (SELECT Max(query)   AS query,
                tbl,
                Max(endtime) AS last_insert
         FROM   system_tables.stl_insert_history
         GROUP  BY tbl),
     max_query_history
     AS (SELECT query,
                querytxt,
                Max(endtime) AS endtime
         FROM   system_tables.stl_query_history
         GROUP  BY query,
                   querytxt),
     final
     AS (SELECT sti.SCHEMA,
                sti.table,
                max_query_history.endtime,
                max_query_history.querytxt
         FROM   max_query_time
                JOIN max_query_history
                  ON max_query_history.query = max_query_time.query
                JOIN svv_table_info sti
                  ON sti.table_id = max_query_time.tbl)
SELECT *
FROM   final 

3. Unlike the BigQuery solution, this test .sql test macro should be defined only once in macros/test_freshness.sql — the solution was inspired by Johann De Wet.

{% test freshness_metadata(model, sch, tbl) %}

WITH max_query_time
     AS (SELECT Max(query)   AS query,
                tbl,
                Max(endtime) AS last_insert
         FROM   system_tables.stl_insert_history
         GROUP  BY tbl),
     max_query_history
     AS (SELECT query,
                querytxt,
                Max(endtime) AS endtime
         FROM   system_tables.stl_query_history
         GROUP  BY query,
                   querytxt),
     final
     AS (SELECT sti.SCHEMA,
                sti.table,
                max_query_history.endtime
         FROM   max_query_time
                JOIN max_query_history
                  ON max_query_history.query = max_query_time.query
                JOIN svv_table_info sti
                  ON sti.table_id = max_query_time.tbl
        where sti.table =  '{{ tbl }}' and sti.SCHEMA = '{{ sch }}')
select datediff(day, date(endtime), current_Date) as days_since_refresh

from final where  datediff(day, date(endtime), current_Date)  >= 1

{% endtest %}

4. Data source properties are defined in .yml file(s) nested under models/<filename>.yml. Specify the arguments — the table name and schema below the test name.

version: 2

sources:
  - name: ticket
    schema: public
    description: Raw ticket data
    tables:
      - name: sales
        description: The table contains raw data on ticket sales
        tests:
          - freshness_metadata:
              tbl: sales
              sch: public
      - name: users
        description: The table contains raw data on users
        tests:
          - freshness_metadata:
              tbl: users
              sch: public

5. Once you are done, run dbt test in your command line. If your data sources were updated less than one day ago, the test will be passed i.e. there will be no results. If a row gets returned by the query, that indicates a failure. sales data was updated today while users' data was updated more than one day ago which was indicated by the test results.

Things to keep in mind:

(1) If you get the permissions error, check the rights that the database user you are using with dbt has. Review https://discourse.getdbt.com/t/the-exact-privileges-we-grant-to-set-up-redshift/2909 as it contains the roles and groups that a dbt user should have to be able to query and modify the existing tables.

(2) If needed, you might be required to grant your user access to the system tables withalter user yourUser syslog access unrestricted;

(3) I am not entirely positive about the table svv_table_info. You might need to stash it in a historical table, too.

(4) Don’t forget to schedule the system tables retrieval.