We previously introduced the need for point-in-time snapshots for an Analytics Engineering team and walked through the basic of setting up a prototype framework for managing them in DBT. We will now extend the prototype with a feature increment (support for backfilling snapshots), improve the robustness of the solution and walk you through considerations regarding deployment & orchestration.
As soon as we shipped the snapshots implementation, the natural question came from adopting stakeholders: This looks nice, we want more. Can we get snapshots since the beginning of this year?
Such an implementation would naturally only work with historized sources (as introduced in Part 1), for which a point-in-time snapshot can be constructed at any arbitrary date in the past.
We designed this feature to rely on declarative configuration (in the DBT underlying model, regarding the earliest date since snapshots should be backfilled if missing) and automated execution (no explicit need to trigger backfilling; it should be executed automatically only if and when needed).
We are extending the execute_snapshot macro from Part 1 to support the configuration:
{% set retrofill_data_since_date = config.get('retrofill_data_since_date') %}
{% set MAX_SCOPES = 30 %}
In the above, retrofill_data_since_date is a parameter which we configure directly in the underlying model in DBT, while MAX_SCOPES is a technical constant intended to limit the number of historical periods (“scopes”) which are backfilled in any single execution and avoid overrunning the execution environment (in our case, the Snowflake virtual warehouse) — see below.
Further, the process determines the “scopes” which need execution (based on the configured start date for backfill and the snapshots which are already created, if any):
{# 3. DETERMINE SCOPE (RETROFILL + CURRENT WORK) #}
{% set query %}
{# make sure that target period is always included, regardless of whether retrofill_data_since_date might hide it #}
SELECT CURRENT_DATE - {{snapshot_delay_days}} as d
UNION ALL
{# ... add backfill scopes, excluding target period from retrofill; already included above #}
SELECT d FROM (
SELECT DATEADD(day, -seq4(), current_date()) as d
FROM TABLE(GENERATOR(ROWCOUNT=>2000))
) WHERE d >= '{{retrofill_data_since_date}}'
AND d < CURRENT_DATE - {{snapshot_delay_days}}
AND d NOT IN (SELECT DISTINCT EFFECTIVE_ON FROM {{this}})
ORDER BY d DESC
{% endset %}
{% set scopes = run_query(query).columns[0].values()[0:MAX_SCOPES] %}
{# print(scopes) #}
In the above, we make sure that the target period snapshot is always rebuilt in a run (another robustness measure, to ensure idempotent execution!) and we add all the missing historical scopes since the beginning of the backfill date (in this case, each scope being one calendar day). By ordering the scopes by descending period and taking the first MAX_SCOPES values, we then restrict the number of scopes as intended.
Finally, the materialization step is modified to determine and persist the snapshot for each scope, in a loop:
{# 4b. MATERIALIZE #}
{% for sc in scopes %}
{% set query %}
INSERT INTO {{this}}
SELECT * FROM TABLE({{this}}('{{sc}}'::DATE))
{% endset %}
{{ log('Materializing snapshot data for ''' ~ sc ~ '''...', True) }}
{% set results = run_query(query) %}
{% endfor %}
Backfilling is not only about retroactive filling of date, yet it also increases the robustness of orchestration because the backfill mechanism would also be used when there are orchestration failures and some executions might be missed and need to be retriggered later. Architecturally, robustness in data pipelines is always something that we aim for, and it more than once spared us late-at-night on-call work.
In order to avoid overrunning the execution environment (in our case, the Snowflake virtual warehouse), as mentioned above, we are dynamically increasing the warehouse size whenever a data backfill needs to be executed. This is determined by the number of scopes, as a normal run would require the building of the current snapshot only (1 scope) while a backfill would require 1+ scopes.
{# 4a. INCREASE WAREHOUSE SIZE IF DATA RETROFILL IS NEEDED #}
{% if scopes|length > 1 %}
{% set query %}
USE WAREHOUSE WH_XL
{% endset %}
{{ log('Setting larger warehouse for execution because data backfill is requested!', True) }}
{% set results = run_query(query) %}
{% else %}
{{ print('Using default WH!') }}
{% endif %}
We started by asking ourselves the following questions and agreeing on the requirements with our stakeholders:
- Would the snapshot need to run less frequently, more frequently or just as frequent as the underlying model?
- How fresh do we need the snapshot data to be?
- How would we align the snapshot run with the underling model run?
On #1, we started on the thought that it does not make sense for the snapshot (for efficiency purposes) to run more often than the underlying model is refreshed. That means that if we want to run an hourly snapshot based on an underlying model which is refreshed daily, then this would simply waste resources as 23 of the 24 executions would simply execute on unchanged data.
On #2, we of course want the snapshot data to be as fresh as possible. One might think — “if I need hourly snapshots at the beginning of each business day, then why don’t I run all snapshots for the last day once, in the morning?”. This would involve a data backfill in the morning based on the scopes missed during the night, yet based on the way we designed our materialization (a series of incremental materialization runs, once for each snapshot in scope) then this is not a tradeoff at all — as running 10 scopes at once (iteratively) or running 10 times 1 scope would consume the same resources. Thus **we chose freshness **— materializing each snapshot as soon as data for it is available and stable.
Finally, on #3 we have a tradeoff between convenience/simplicity and fit-for-purpose.
Convenience means that we want to avoid creating an additional orchestration for the snapshot, separate from the one of the underlying model. If the underlying model is refreshed daily and we have a daily snapshot too, then this is a no-brainer — the two models should be materialized in the same data pipeline, as the underlying model is simply an upstream dependency of our snapshot model. This is our case as well.
On the other hand, if we have a daily-refreshed underlying model and we want an hourly snapshot, then co-orchestration would mean making an explicit tradeoff for convenience (backfill the snapshots for the last 24 hours only once per day, triggered by the execution of the underlying model) to the detriment of snapshot freshness. This is also related to #2 above.
There are at least two aspects we are aware about and which we intentionally didn’t bake into our design:
- schema evolution — in case the schema of the underlying model changes, the snapshot model would need to change too (as the materialization would simply fail otherwise due to the select * query); the automation (macro) does not handle such scenarios out of the box, and the best workaround is to drop the snapshot table completely and allow the macro to backfill the data since the declared start date;
- logic evolution — similar to the above, in case of any logic changes that would impact the snapshots (or bugfixes), the snapshots would need to be recreated.
The macro is now reworked to incorporate the improvements regarding backfilling and orchestration:
{#
Manage the materialization of a periodic snapshot for an existing model.
With support for robust data backfilling.
#}
{% macro execute_snapshot(this) -%}
{% if execute %}
{% set retrofill_data_since_date = config.get('retrofill_data_since_date') %}
{% set snapshot_delay_days = config.get('snapshot_delay_days', default = 2) %}
{% set MAX_SCOPES = 30 %}
{# 1. CREATE INFRASTRUCTURE (IDEMPOTENT) #}
{% set query %}
SELECT TRIM(REPLACE(REPLACE(DATA_TYPE, 'TABLE (', ''), ')', ''))
FROM {{this.database}}.INFORMATION_SCHEMA.FUNCTIONS
WHERE function_schema = UPPER('{{this.schema}}')
AND function_name = UPPER('{{this.identifier}}')
{% endset %}
{# print(query) #}
{% set results = run_query(query) %}
{% set funcsig = results.columns[0].values()[0] %}
{% set query %}
CREATE TRANSIENT TABLE IF NOT EXISTS {{this}} ({{funcsig}})
{% endset %}
{# print(query) #}
{% set results = run_query(query) %}
{# 2. CLEANUP CURRENT SNAPSHOT (IT SHOULD NOT EXIST IF SNAPSHOT IS TRIGGER ONCE PER PERIOD) #}
{% set query %}
DELETE FROM {{this}} WHERE EFFECTIVE_ON >= CURRENT_DATE - {{snapshot_delay_days}}
{% endset %}
{# print(query) #}
{# set results = run_query(query) #}
{# 3. DETERMINE SCOPE (RETROFILL + CURRENT WORK) #}
{% set query %}
{# make sure that target period is always included, regardless of whether retrofill_data_since_date might hide it... #}
SELECT CURRENT_DATE - {{snapshot_delay_days}} as d
UNION ALL
{# ... add backfill scopes, excluding target period from retrofill; already included above #}
SELECT d FROM (
SELECT DATEADD(day, -seq4(), current_date()) as d
FROM TABLE(GENERATOR(ROWCOUNT=>2000))
) WHERE d >= '{{retrofill_data_since_date}}'
AND d < CURRENT_DATE - {{snapshot_delay_days}}
AND d NOT IN (SELECT DISTINCT EFFECTIVE_ON FROM {{this}})
ORDER BY d DESC
{% endset %}
{% set scopes = run_query(query).columns[0].values()[0:MAX_SCOPES] %}
{# print(scopes) #}
{# 4a. INCREASE WAREHOUSE SIZE IF DATA RETROFILL IS NEEDED #}
{% if scopes|length > 1 %}
{% set query %}
USE WAREHOUSE WH_XL
{% endset %}
{{ log('Setting larger warehouse for execution because data backfill is requested!', True) }}
{% set results = run_query(query) %}
{% else %}
{{ print('Using default WH!') }}
{% endif %}
{# 4b. MATERIALIZE #}
{% for sc in scopes %}
{% set query %}
INSERT INTO {{this}}
SELECT * FROM TABLE({{this}}('{{sc}}'::DATE))
{% endset %}
{{ log('Materializing snapshot data for ''' ~ sc ~ '''...', True) }}
{% set results = run_query(query) %}
{% endfor %}
{% endif %}
{%- endmacro %}
Invoking the macro in the underlying model is done by adding the following DBT configuration for it:
{{ config(
post_hook = [
"{{ execute_snapshot(this) }}"
],
snapshot_delay_days = 2,
retrofill_data_since_date = '2024-01-01'
)}}
Designing data pipelines is a complex exercise that is balanced among two forces — on one side, satisfying the requests of your downstream consumers, and on the other side, a top-down organizational/governance driver to keep a grip on the complexity and defining some reusable templates & patterns that have a larger applicability than one specific use-case.
In our case, we managed to incorporate into the AE development flow the technical aspect of exposing data in point-in-time snapshot format and some design considerations that the underlying models need to satisfy. This provides predictability, consistency and easy of maintenance for the AE team while it allows for downstream teams to build further data modeling constructs starting from a clean foundation of managed, data-quality-proofed and easy-to-query datasets.
A related approach to ours, dimensional snapshotting, is mentioned here as part of the functional data engineering methodology proposed by Preset. They share some potentially interesting add-ons to consider, like improving the query efficiency by partitioning the snapshot table by calendar date or facilitating consumption by exposing a dedicated view with the latest snapshot of an entity.Also published on Medium
