Scaling Statistics: Incremental Commonplace Deviation in SQL with dbt | by Yuval Gorchover | Jan, 2025

SQL aggregation capabilities could be computationally costly when utilized to giant datasets. As datasets develop, recalculating metrics over the whole dataset repeatedly turns into inefficient. To deal with this problem, incremental aggregation is commonly employed — a way that includes sustaining a earlier state and updating it with new incoming knowledge. Whereas this method is easy for aggregations like COUNT or SUM, the query arises: how can it’s utilized to extra complicated metrics like commonplace deviation?

Commonplace deviation is a statistical metric that measures the extent of variation or dispersion in a variable’s values relative to its imply.
It’s derived by taking the sq. root of the variance.
The components for calculating the variance of a pattern is as follows:

Calculating commonplace deviation could be complicated, because it includes updating each the imply and the sum of squared variations throughout all knowledge factors. Nonetheless, with algebraic manipulation, we will derive a components for incremental computation — enabling updates utilizing an present dataset and incorporating new knowledge seamlessly. This method avoids recalculating from scratch every time new knowledge is added, making the method rather more environment friendly (An in depth derivation is obtainable on my GitHub).

The components was mainly damaged into 3 components:
1. The prevailing’s set weighted variance
2. The brand new set’s weighted variance
3. The imply distinction variance, accounting for between-group variance.

This methodology permits incremental variance computation by retaining the COUNT (ok), AVG (µk), and VAR (Sk) of the prevailing set, and mixing them with the COUNT (n), AVG (µn), and VAR (Sn) of the brand new set. Consequently, the up to date commonplace deviation could be calculated effectively with out rescanning the whole dataset.

Now that we’ve wrapped our heads across the math behind incremental commonplace deviation (or at the least caught the gist of it), let’s dive into the dbt SQL implementation. Within the following instance, we’ll stroll via find out how to arrange an incremental mannequin to calculate and replace these statistics for a person’s transaction knowledge.

Take into account a transactions desk named stg__transactions, which tracks person transactions (occasions). Our purpose is to create a time-static desk, int__user_tx_state, that aggregates the ‘state’ of person transactions. The column particulars for each tables are supplied within the image under.

To make the method environment friendly, we purpose to replace the state desk incrementally by combining the brand new incoming transactions knowledge with the prevailing aggregated knowledge (i.e. the present person state). This method permits us to calculate the up to date person state with out scanning via all historic knowledge.

The code under assumes understanding of some dbt ideas, when you’re unfamiliar with it, you should still be capable to perceive the code, though I strongly encourage going via dbt’s incremental information or learn this superior publish.

We’ll assemble a full dbt SQL step-by-step, aiming to calculate incremental aggregations effectively with out repeatedly scanning the whole desk. The method begins by defining the mannequin as incremental in dbt and utilizing unique_key to replace present rows reasonably than inserting new ones.

-- depends_on: {{ ref('stg__transactions') }}
{{ config(materialized='incremental', unique_key=['USER_ID'], incremental_strategy='merge') }}

Subsequent, we fetch information from the stg__transactions desk.
The is_incremental block filters transactions with timestamps later than the most recent person replace, successfully together with “solely new transactions”.

WITH NEW_USER_TX_DATA AS (
SELECT
USER_ID,
TX_ID,
TX_TIMESTAMP,
TX_VALUE
FROM {{ ref('stg__transactions') }}
{% if is_incremental() %}
WHERE TX_TIMESTAMP > COALESCE((choose max(UPDATED_AT) from {{ this }}), 0::TIMESTAMP_NTZ)
{% endif %}
)

After retrieving the brand new transaction information, we combination them by person, permitting us to incrementally replace every person’s state within the following CTEs.

INCREMENTAL_USER_TX_DATA AS (
SELECT
USER_ID,
MAX(TX_TIMESTAMP) AS UPDATED_AT,
COUNT(TX_VALUE) AS INCREMENTAL_COUNT,
AVG(TX_VALUE) AS INCREMENTAL_AVG,
SUM(TX_VALUE) AS INCREMENTAL_SUM,
COALESCE(STDDEV(TX_VALUE), 0) AS INCREMENTAL_STDDEV,
FROM
NEW_USER_TX_DATA
GROUP BY
USER_ID
)

Now we get to the heavy half the place we have to truly calculate the aggregations. After we’re not in incremental mode (i.e. we don’t have any “state” rows but) we merely choose the brand new aggregations

NEW_USER_CULMULATIVE_DATA AS (
SELECT
NEW_DATA.USER_ID,
{% if not is_incremental() %}
NEW_DATA.UPDATED_AT AS UPDATED_AT,
NEW_DATA.INCREMENTAL_COUNT AS COUNT_TX,
NEW_DATA.INCREMENTAL_AVG AS AVG_TX,
NEW_DATA.INCREMENTAL_SUM AS SUM_TX,
NEW_DATA.INCREMENTAL_STDDEV AS STDDEV_TX
{% else %}
...

However after we’re in incremental mode, we have to be part of previous knowledge and mix it with the brand new knowledge we created within the INCREMENTAL_USER_TX_DATA CTE primarily based on the components described above.
We begin by calculating the brand new SUM, COUNT and AVG:

  ...
{% else %}
COALESCE(EXISTING_USER_DATA.COUNT_TX, 0) AS _n, -- that is n
NEW_DATA.INCREMENTAL_COUNT AS _k,  -- that is ok
COALESCE(EXISTING_USER_DATA.SUM_TX, 0) + NEW_DATA.INCREMENTAL_SUM AS NEW_SUM_TX,  -- new sum
COALESCE(EXISTING_USER_DATA.COUNT_TX, 0) + NEW_DATA.INCREMENTAL_COUNT AS NEW_COUNT_TX,  -- new rely
NEW_SUM_TX / NEW_COUNT_TX AS AVG_TX,  -- new avg
...

We then calculate the variance components’s three components

1. The prevailing weighted variance, which is truncated to 0 if the earlier set consists of 1 or much less objects:

    ...
CASE
WHEN _n > 1 THEN (((_n - 1) / (NEW_COUNT_TX - 1)) * POWER(COALESCE(EXISTING_USER_DATA.STDDEV_TX, 0), 2))
ELSE 0
END AS EXISTING_WEIGHTED_VARIANCE,  -- present weighted variance
...

2. The incremental weighted variance in the identical method:

    ...
CASE
WHEN _k > 1 THEN (((_k - 1) / (NEW_COUNT_TX - 1)) * POWER(NEW_DATA.INCREMENTAL_STDDEV, 2))
ELSE 0
END AS INCREMENTAL_WEIGHTED_VARIANCE,  -- incremental weighted variance
...

3. The imply distinction variance, as outlined earlier, together with SQL be part of phrases to incorporate previous knowledge.

    ...
POWER((COALESCE(EXISTING_USER_DATA.AVG_TX, 0) - NEW_DATA.INCREMENTAL_AVG), 2) AS MEAN_DIFF_SQUARED,
CASE
WHEN NEW_COUNT_TX = 1 THEN 0
ELSE (_n * _k) / (NEW_COUNT_TX * (NEW_COUNT_TX - 1))
END AS BETWEEN_GROUP_WEIGHT,  -- between group weight
BETWEEN_GROUP_WEIGHT * MEAN_DIFF_SQUARED AS MEAN_DIFF_VARIANCE,  -- imply diff variance
EXISTING_WEIGHTED_VARIANCE + INCREMENTAL_WEIGHTED_VARIANCE + MEAN_DIFF_VARIANCE AS VARIANCE_TX,
CASE
WHEN _n = 0 THEN NEW_DATA.INCREMENTAL_STDDEV -- no "previous" knowledge
WHEN _k = 0 THEN EXISTING_USER_DATA.STDDEV_TX -- no "new" knowledge
ELSE SQRT(VARIANCE_TX)  -- stddev (which is the basis of variance)
END AS STDDEV_TX,
NEW_DATA.UPDATED_AT AS UPDATED_AT,
NEW_SUM_TX AS SUM_TX,
NEW_COUNT_TX AS COUNT_TX
{% endif %}
FROM
INCREMENTAL_USER_TX_DATA new_data
{% if is_incremental() %}
LEFT JOIN
{{ this }} EXISTING_USER_DATA
ON
NEW_DATA.USER_ID = EXISTING_USER_DATA.USER_ID
{% endif %}
)

Lastly, we choose the desk’s columns, accounting for each incremental and non-incremental circumstances:

SELECT
USER_ID,
UPDATED_AT,
COUNT_TX,
SUM_TX,
AVG_TX,
STDDEV_TX
FROM NEW_USER_CULMULATIVE_DATA

By combining all these steps, we arrive on the remaining SQL mannequin:

-- depends_on: {{ ref('stg__initial_table') }}
{{ config(materialized='incremental', unique_key=['USER_ID'], incremental_strategy='merge') }}
WITH NEW_USER_TX_DATA AS (
SELECT
USER_ID,
TX_ID,
TX_TIMESTAMP,
TX_VALUE
FROM {{ ref('stg__initial_table') }}
{% if is_incremental() %}
WHERE TX_TIMESTAMP > COALESCE((choose max(UPDATED_AT) from {{ this }}), 0::TIMESTAMP_NTZ)
{% endif %}
),
INCREMENTAL_USER_TX_DATA AS (
SELECT
USER_ID,
MAX(TX_TIMESTAMP) AS UPDATED_AT,
COUNT(TX_VALUE) AS INCREMENTAL_COUNT,
AVG(TX_VALUE) AS INCREMENTAL_AVG,
SUM(TX_VALUE) AS INCREMENTAL_SUM,
COALESCE(STDDEV(TX_VALUE), 0) AS INCREMENTAL_STDDEV,
FROM
NEW_USER_TX_DATA
GROUP BY
USER_ID
),
NEW_USER_CULMULATIVE_DATA AS (
SELECT
NEW_DATA.USER_ID,
{% if not is_incremental() %}
NEW_DATA.UPDATED_AT AS UPDATED_AT,
NEW_DATA.INCREMENTAL_COUNT AS COUNT_TX,
NEW_DATA.INCREMENTAL_AVG AS AVG_TX,
NEW_DATA.INCREMENTAL_SUM AS SUM_TX,
NEW_DATA.INCREMENTAL_STDDEV AS STDDEV_TX
{% else %}
COALESCE(EXISTING_USER_DATA.COUNT_TX, 0) AS _n, -- that is n
NEW_DATA.INCREMENTAL_COUNT AS _k,  -- that is ok
COALESCE(EXISTING_USER_DATA.SUM_TX, 0) + NEW_DATA.INCREMENTAL_SUM AS NEW_SUM_TX,  -- new sum
COALESCE(EXISTING_USER_DATA.COUNT_TX, 0) + NEW_DATA.INCREMENTAL_COUNT AS NEW_COUNT_TX,  -- new rely
NEW_SUM_TX / NEW_COUNT_TX AS AVG_TX,  -- new avg
CASE
WHEN _n > 1 THEN (((_n - 1) / (NEW_COUNT_TX - 1)) * POWER(COALESCE(EXISTING_USER_DATA.STDDEV_TX, 0), 2))
ELSE 0
END AS EXISTING_WEIGHTED_VARIANCE,  -- present weighted variance
CASE
WHEN _k > 1 THEN (((_k - 1) / (NEW_COUNT_TX - 1)) * POWER(NEW_DATA.INCREMENTAL_STDDEV, 2))
ELSE 0
END AS INCREMENTAL_WEIGHTED_VARIANCE,  -- incremental weighted variance
POWER((COALESCE(EXISTING_USER_DATA.AVG_TX, 0) - NEW_DATA.INCREMENTAL_AVG), 2) AS MEAN_DIFF_SQUARED,
CASE
WHEN NEW_COUNT_TX = 1 THEN 0
ELSE (_n * _k) / (NEW_COUNT_TX * (NEW_COUNT_TX - 1))
END AS BETWEEN_GROUP_WEIGHT,  -- between group weight
BETWEEN_GROUP_WEIGHT * MEAN_DIFF_SQUARED AS MEAN_DIFF_VARIANCE,
EXISTING_WEIGHTED_VARIANCE + INCREMENTAL_WEIGHTED_VARIANCE + MEAN_DIFF_VARIANCE AS VARIANCE_TX,
CASE
WHEN _n = 0 THEN NEW_DATA.INCREMENTAL_STDDEV -- no "previous" knowledge
WHEN _k = 0 THEN EXISTING_USER_DATA.STDDEV_TX -- no "new" knowledge
ELSE SQRT(VARIANCE_TX)  -- stddev (which is the basis of variance)
END AS STDDEV_TX,
NEW_DATA.UPDATED_AT AS UPDATED_AT,
NEW_SUM_TX AS SUM_TX,
NEW_COUNT_TX AS COUNT_TX
{% endif %}
FROM
INCREMENTAL_USER_TX_DATA new_data
{% if is_incremental() %}
LEFT JOIN
{{ this }} EXISTING_USER_DATA
ON
NEW_DATA.USER_ID = EXISTING_USER_DATA.USER_ID
{% endif %}
)
SELECT
USER_ID,
UPDATED_AT,
COUNT_TX,
SUM_TX,
AVG_TX,
STDDEV_TX
FROM NEW_USER_CULMULATIVE_DATA

All through this course of, we demonstrated find out how to deal with each non-incremental and incremental modes successfully, leveraging mathematical strategies to replace metrics like variance and commonplace deviation effectively. By combining historic and new knowledge seamlessly, we achieved an optimized, scalable method for real-time knowledge aggregation.

On this article, we explored the mathematical method for incrementally calculating commonplace deviation and find out how to implement it utilizing dbt’s incremental fashions. This method proves to be extremely environment friendly, enabling the processing of huge datasets with out the necessity to re-scan the whole dataset. In apply, this results in quicker, extra scalable techniques that may deal with real-time updates effectively. If you happen to’d like to debate this additional or share your ideas, be at liberty to achieve out — I’d love to listen to your ideas!