Desining and Building an Online Retail Data Warehouse with dbt and Snowflake

This article is intended for data engineers with a fundamental grasp of the following principles:

• Python programming.
• SQL querying.
• Extract, Load, Transform (ELT) pipelines.
• Star schema data modeling.
• Data warehousing.
• Data validation and cleansing.

I design and implement a data warehouse following an ELT paradigm with the following technology stack:

• Python for data ingestion.
• Snowflake for data storage.
• Dbt for data transformation.

The article does not cover pipeline orchestration or other topics outside this scope.

Online Retail Data Warehouse with dbt and Snowflake

Implementation Roadmap

The following roadmap outlines the implementation process:

Data warehouse
├── Dataset analysis
│   ├── Analytical requirements
│   ├── Data quality requirements
│   └── Define the grain
├── Data modeling
│   ├── Conceptual model
│   ├── Logical model
│   └── Physical model
├── Architecture overview
├── Data ingestion
├── dbt setup
│   ├── Environment setup
│   └── Create project and connect to Snowflake
└── Data quality management
    ├── Staging
    │   ├── Data validation
    │   │   ├── Data type check
    │   │   └── Data uniqueness check
    │   └── Data cleansing
    │       └── Type cast and Column rename
    ├── Intermediate
    │   ├── Data validation
    │   │   ├── Range check
    │   │   └── Format check
    │   └── Data cleansing
    │       ├── Deduplication
    │       ├── Missing values
    │       └── Derived fields
    └── Marts
        ├── Implement dimensions
        └── Implement fact table

Dataset Analysis

I use the Online Retail dataset by Daqing Chen, made available through the UCI Machine Learning Repository under the CC BY 4.0 license.

This dataset represents transactions from a UK-based and registered non-store online retailer. The dataset comprises 541,909 transaction records, recorded between 1 December 2010 and 9 December 2011.

The dataset contains the following features:

• InvoiceNo: Six-digit numbers that uniquely identify transactions; codes starting with C indicate cancellations.
• StockCode: Codes that uniquely identify products.
• Description: Product names.
• Quantity: Quantity of a product in each transaction.
• InvoiceDate: Date and time when the transaction was generated.
• UnitPrice: Product price per unit.
• CustomerID: Five-digit numbers that uniquely identify customers.
• Country: Country where the customers reside.

Analytical requirements

An analysis of the dataset revealed several opportunities that can be addressed through a structured data warehouse. These opportunities are primarily centered around cancelled transactions. Based on these observations, I outline the following analytical questions that the data warehouse should answer:

• Which customers generate the most cancelled transactions?
• Which customers generate the most transactions while having the lowest cancellation rate?
• Which customers generate the most revenue?
• Which customers purchase the largest quantity of products?
• Which customers purchase products with the highest average price per unit?
• Which customers generate high revenue while maintaining a low cancellation rate?
• Which periods experience the highest number of cancellations?
• How does the cancellation rate vary throughout the year?
• Which products are most frequently associated with cancellations?
• Which countries have the highest cancellation rate, and which have the lowest?

Data quality requirements

This analysis also surfaced the following data quality issues:

• Missing values.
• Negative unit prices.
• Duplicate records.

Negative quantities may represent cancelled/returned transactions rather than ordinary sales. Negative unit price, on the other hand, may indicate anomalous values requiring further investigation.

Define the grain

The examination of the dataset indicates that a single transaction can span multiple rows since a transaction may contain multiple products.

Figure 1 provides an example from the dataset.

Figure 1. Example of a transaction within the Online Retail dataset

Grain: One row represents one transaction record for a product and customer.

Data modeling

Data modeling is the translation of business requirements into organized data structures that support analysis. I design the data model following the three phases:

• Conceptual model.
• Logical model.
• Physical model.

Conceptual model

The conceptual data model provides a high-level representation of the main business entities and the relationships between them. I do not implement details such as columns, data types, and database-specific structures at this stage.

The main entities identified from the analytical requirements are:

• Transaction.
• Product.
• Customer.

Country is considered an attribute of Customer and not as a separate entity.

Logical model

The logical model offers a detailed description of the data and expands on the conceptual model.

At this stage, I choose the star schema for the data warehouse.

The fact table is connected to dimensions using foreign keys and is depicted in the center.

Dimensions

• dim_customer (Primary key: customer_key, surrogate) - customer_id, country.
• dim_product (Primary key: product_key, surrogate) - stock_code, description.
• dim_date (Primary key: date_key, surrogate) - invoice_date, i_year, i_month, i_day.

Fact table: fact_transaction

• Grain: One row represents one transaction record for a product and customer.

• Primary key: transaction_key, surrogate.

• Measures: quantity, unit_price.

• Degenerate key: invoice_id.

• Foreign keys: customer_key, product_key, date_key.

• Derived fields: is_cancelled (determines whether the transaction is cancelled or not). total_cost (multiply quantity by unit_price).

Relationships

• dim_customer 1 → N fact_transaction.
• dim_product 1 → N fact_transaction.
• dim_date 1 → N fact_transaction.

Physical model

The physical model represents the implementation of the data structure in a specific database management system. In this project, I choose Snowflake as the data warehouse.

Figure 2 illustrates the physical implementation of the data warehouse schema.

Figure 2. Physical model of the Online Retail data warehouse in Snowflake.

Architecture overview

Figure 3 illustrates the pipeline's system architecture.

Figure 3. Online retail data warehouse pipeline architecture. A Python application (extract and load modules, using snowflake-connector-python) is used to extract and load raw sales data (online_retail.xlsx) into Snowflake. The staging, intermediate, and marts layers of a staged pipeline are then used by dbt to modify the raw data, creating tables in the Snowflake data warehouse (online_retail_warehouse) that are ready for analytics.

Data ingestion

I create a Python application for the data ingestion layer. The following tree represents the project structure:

online_retail_ingestion/
│
├── data/
│   └── online_retail.xlsx
│
├── workflow/
│   ├── __init__.py
│   ├── extract.py
│   └── load.py
│
├── main.py
├── requirements.txt
├── .gitignore
└── README.md

The dataset is stored in the data folder, while the extraction and loading stages are implemented in separate modules.

The application workflow consists of three steps:

• Extract the data from the .xlsx file using the OpenPyXL library.
• Establish a connection to Snowflake.
• Load raw data into a Snowflake table.

See GitHub repository for full implementation.

Connect to Snowflake

To connect the Python application to Snowflake, I need the following configuration values:

• Warehouse.
• Database.
• Schema.

First, I created a Snowflake free trial account, and selected Google Cloud Platform as the cloud data storage for my Snowflake environment.

Using the Snowflake web interface Snowsight, I navigate to the workspace and create a new SQL file. I use this file to provision the resources required by the ingestion layer. The following SQL statements create the warehouse, the database, and the schema:

CREATE WAREHOUSE online_retail_warehouse
WAREHOUSE_SIZE = 'XSMALL'
AUTO_SUSPEND = 900; 

USE WAREHOUSE online_retail_warehouse;
CREATE DATABASE online_retail_db;

USE DATABASE online_retail_db;
CREATE SCHEMA retail;

I set the size of the warehouse. AUTO_SUSPEND = 900 suspends the warehouse after 900 seconds (15 minutes) of inactivity.

Once I specify these configuration values, I use the snowflake-connector-python library to establish the connection. I load the raw data inside the invoices_raw_data table in the retail schema.

dbt setup

Environment setup

Connecting dbt to Snowflake requires a set of configuration values.

In Snowsight, I navigate to the SQL file in the workspace and execute the following SQL statements:

CREATE ROLE dbt_role;
GRANT USAGE ON WAREHOUSE online_retail_warehouse TO ROLE dbt_role;
GRANT USAGE ON DATABASE online_retail_db TO ROLE dbt_role;
GRANT USAGE, CREATE SCHEMA ON DATABASE online_retail_db TO ROLE dbt_role;
GRANT ALL PRIVILEGES ON SCHEMA ONLINE_RETAIL_DB.RETAIL TO ROLE dbt_role;

CREATE USER your_username PASSWORD='your_password'
    DEFAULT_ROLE = dbt_role
    DEFAULT_WAREHOUSE = online_retail_warehouse
    DEFAULT_NAMESPACE = ONLINE_RETAIL_DB.RETAIL;

GRANT ROLE dbt_role TO USER your_username

In a separate Python virtual environment, I install dbt-core and dbt-snowflake using:

pip install dbt-core dbt-snowflake

Once the packages are installed, I create the profiles.yml file. This YAML configuration file defines the Snowflake profile used by dbt.

On Windows, I navigate to the user folder and create the .dbt directory. Inside this directory, I create profiles.yml as follows:

snowflake-db:
  target: dev

  outputs:
    dev:
      type: snowflake
      account: [your Snowflake account identifier]
      user: [username chosen in Snowflake]
      password: [password chosen in Snowflake]
      role: dbt_role
      warehouse: online_retail_warehouse
      database: online_retail_db
      schema: RETAIL
      threads: 4
      client_session_keep_alive: False
      query_tag: [define a query tag, ex: dbt_online_retail]

Create project and connect to Snowflake

Once the configuration requirements are complete, I initialize the dbt project inside the Python virtual environment. The following command initializes the dbt project using the snowflake-db profile.

dbt init --profile snowflake-db

This command prompts me to submit the project name. In this case, the project name is online_retail_dbt. The command creates the standard dbt project scaffold.

See the GitHub repository for the full implementation of the transformation layer.

Inside dbt_project.yml, I set the profile value to snowflake-db which references the profile created in profiles.yml:

name: 'online_retail_dbt'
version: '1.0.0'

# This setting configures which "profile" dbt uses for this project.
profile: 'snowflake-db'

After I save the file, I test the connection by running:

dbt debug --connection

Data quality management

In the dbt_project.yml file, I define how the models in each transformation layer are materialized in the warehouse:

models:
  online_retail_dbt:

    staging:
      +materialized: view

    intermediate:
      +materialized: view

    marts:
      +materialized: table

Staging

I create the staging subdirectory in the models directory. Next, I add the sources.yml file. This file declares the raw Snowflake table as a dbt source:

version: 2

sources:
  - name: raw
    database: ONLINE_RETAIL_DB
    schema: RETAIL

    tables:
      - name: invoices_raw_data

Next, I create the schema.yml to document the staging model and its columns:

version: 2

models:

    - name: stg_online_retail

      columns:
        - name: invoice_id

        - name: stock_code

        - name: description

        - name: quantity

        - name: invoice_date

        - name: unit_price

        - name: customer_id

        - name: country

Data validation

The data validation files are created inside the tests directory.

At this stage, I perform the following checks:

• dq_data_type_check.sql: validates the data types of all the columns in the invoices_raw_data table.
• dq_uniqueness_check.sql: validates the uniqueness of the combination of (INVOICENO, STOCKCODE and CUSTOMERID) in the invoices_raw_data table.

Data type check

I create the dq_data_type_check.sql file as follows:

WITH
  expected_types AS (
    SELECT
      'INVOICENO' AS COLUMN_NAME,
      'VARCHAR' AS EXPECTED_DATA_TYPE
    UNION ALL
    SELECT
      'STOCKCODE',
      'VARCHAR'
    UNION ALL
    SELECT
      'DESCRIPTION',
      'VARCHAR'
    -- Define the expected data type for each column in the raw table.
  ),
  actual_types AS (
    SELECT
      COLUMN_NAME,
      DATA_TYPE AS ACTUAL_DATA_TYPE
    FROM
      INFORMATION_SCHEMA.COLUMNS
    WHERE
      TABLE_SCHEMA = 'RETAIL'
      AND TABLE_NAME = 'INVOICES_RAW_DATA'
  )
SELECT
  e.COLUMN_NAME,
  e.EXPECTED_DATA_TYPE,
  a.ACTUAL_DATA_TYPE,
  CASE
    WHEN a.ACTUAL_DATA_TYPE IS NULL THEN 'MISSING'
    WHEN e.EXPECTED_DATA_TYPE = a.ACTUAL_DATA_TYPE THEN 'VALIDATED'
    WHEN e.EXPECTED_DATA_TYPE != a.ACTUAL_DATA_TYPE THEN 'TYPE MISMATCH'
  END AS STATUS
FROM
  expected_types e
  LEFT JOIN actual_types a ON e.COLUMN_NAME = a.COLUMN_NAME
WHERE
  a.ACTUAL_DATA_TYPE IS NULL
  OR e.EXPECTED_DATA_TYPE != a.ACTUAL_DATA_TYPE

This query checks the data types of the columns in the invoices_raw_data table. I specify the expected data types based on the physical model and compare them against the data types in the corresponding columns in invoices_raw_data.

Uniqueness check

I add the dq_uniqueness_check.sql file as follows:

SELECT INVOICENO, STOCKCODE, CUSTOMERID, COUNT(*) AS occurrence_count
FROM {{source("raw", 'invoices_raw_data')}}
GROUP BY INVOICENO, STOCKCODE, CUSTOMERID
HAVING COUNT(*) > 1

This test verifies the occurrence of the combination of INVOICENO, STOCKCODE and CUSTOMERID.

To run the tests, I use the command

dbt test --select <file name>

The uniqueness check indicates that the combination of INVOICENO, STOCKCODE, and CUSTOMERID is not unique. Further investigation showed that repeated combinations can contain different attributes such as quantities or unit prices. Therefore, I don't consider these records duplicates, as this may reflect how the source transaction system records item-level transactions.

Data cleansing

Type cast and Column rename

Inside the staging directory, I create the stg_online_retail.sql file:

SELECT
    CAST(INVOICENO AS VARCHAR) AS invoice_id,
    CAST(STOCKCODE AS VARCHAR) AS stock_code,
    CAST(DESCRIPTION AS VARCHAR) AS description,
    CAST(QUANTITY AS INTEGER) AS quantity,
    CAST(INVOICEDATE AS TIMESTAMP_NTZ) AS invoice_date,
    CAST(UNITPRICE AS NUMBER(10,2)) AS unit_price,
    CAST(CUSTOMERID AS INTEGER) AS customer_id,
    CAST(COUNTRY AS VARCHAR) AS country
FROM {{ source('raw', 'invoices_raw_data') }}

This statement casts each column in invoices_raw_data to its expected data type, and renames the columns according to the physical model.

Intermediate

Data validation

Range check

I create dq_range_check.sql inside the tests directory:

WITH staged_data AS (
    SELECT *
    FROM {{ ref('stg_online_retail') }}
),

range_validation AS(
    SELECT *
    FROM staged_data
    WHERE invoice_date >= '2011-12-10'
        OR invoice_date < '2010-12-01'
        OR unit_price < 0
        OR quantity < 0
)

SELECT *
FROM range_validation

The test indicates the following:

• Negative quantities are only associated with canceled transactions. Therefore, I don't consider this as a quality issue.

• Negative unit prices indicate a quality issue that needs to be addressed.

• All invoices dates fall within the expected date range.

Format check

I create dq_format_check.sql:

WITH staged_data AS (
    SELECT *
    FROM {{ ref('stg_online_retail') }}
),

format_validation AS (
    SELECT invoice_id, quantity, unit_price
    FROM staged_data
    WHERE NOT REGEXP_LIKE(invoice_id, '^(C|[0-9]).*')
        OR NOT REGEXP_LIKE(unit_price, '^-?[0-9]+(\.[0-9]+)?$')
        OR NOT REGEXP_LIKE(quantity, '^-?[0-9]+$')
)

SELECT *
FROM format_validation

The test result indicates three invoice IDs beginning with the letter 'A'. However, these records represent bad-debt adjustments rather than standard retail transactions. Therefore, I exclude them for the next stage.

Data cleansing

Inside the intermediate directory, I create the schema.yml as follows:

version: 2

models:

    - name: int_online_retail

      columns:
        - name: invoice_id

        - name: stock_code

        - name: description

        - name: quantity

        - name: invoice_date

        - name: unit_price

        - name: customer_id

        - name: country

        - name: total_cost

        - name: is_cancelled

Then, I create int_online_retail.sql to create the view:

WITH
  staged_data AS (
    SELECT
      *
    FROM
      {{ ref('stg_online_retail') }}
  ),
  clean_data AS (
    SELECT DISTINCT
      invoice_id,
      stock_code,
      COALESCE(description, 'Unknown') AS description,
      quantity,
      invoice_date,
      ABS(unit_price) AS unit_price,
      COALESCE(customer_id, 0) AS customer_id,
      COALESCE(country, 'Unknown') AS country,
      quantity * ABS(unit_price) AS total_cost,
      CASE
        WHEN invoice_id LIKE 'C%' THEN true
        ELSE false
      END AS is_cancelled
    FROM
      staged_data
    WHERE
      invoice_id NOT LIKE 'A%'
  )
SELECT
  *
FROM
  clean_data

These SQL statements perform the following actions:

• Deduplication: remove only identical rows across all columns.
• Handle missing values: replace missing values with their corresponding placeholder values.
• Create derived fields: Add the total_cost field by multiplying quantity by unit_price. Additionally, negative unit_price values are converted into positive values. Add the boolean is_cancelled field that determines if a transaction is cancelled or not.
• Bad-debt adjustments: Exclude records with invoice IDs starting with "A" from the int_online_retail view.

Marts

At this final stage, I implement the star schema using int_online_retail.

Implement dimensions

I create the following dimensions each with its own surrogate key:

• dim_customer
• dim_product
• dim_date

dim_customer:

WITH
  clean_data AS (
    SELECT
      *
    FROM
      {{ref('int_online_retail') }}
  ),
  dim_customer_nsk AS (
    SELECT
      DISTINCT customer_id,
      country
    FROM
      clean_data
  ),
  dim_customer AS (
    SELECT
      ROW_NUMBER() OVER (
        ORDER BY
          customer_id
      ) AS customer_key,
      customer_id,
      country
    FROM
      dim_customer_nsk
  )
SELECT
  *
FROM
  dim_customer

dim_product:

WITH
  clean_data AS (
    SELECT
      *
    FROM
      {{ref('int_online_retail') }}
  ),
  dim_product_nsk AS (
    SELECT
      DISTINCT stock_code,
      description
    FROM
      clean_data
  ),
  dim_product AS (
    SELECT
      ROW_NUMBER() OVER(
        ORDER BY
          stock_code
      ) AS product_key,
      stock_code,
      description
    FROM
      dim_product_nsk
  )
SELECT
  *
FROM
  dim_product

dim_date:

WITH
  clean_data AS (
    SELECT
      *
    FROM
      {{ref('int_online_retail') }}
  ),
  dim_date_nsk AS (
    SELECT
      DISTINCT invoice_date,
      EXTRACT(
        YEAR
        FROM
          invoice_date
      ) AS i_year,
      EXTRACT(
        MONTH
        FROM
          invoice_date
      ) AS i_month,
      EXTRACT(
        DAY
        FROM
          invoice_date
      ) AS i_day
    FROM
      clean_data
  ),
  dim_date AS (
    SELECT
      ROW_NUMBER() OVER (
        ORDER BY
          invoice_date
      ) AS date_key,
      invoice_date,
      i_year,
      i_month,
      i_day
    FROM
      dim_date_nsk
  )
SELECT
  *
FROM
  dim_date

Implement fact table

I create fact_transaction.sql inside the marts directory:

SELECT
    ROW_NUMBER() OVER (
        ORDER BY t.invoice_id, t.stock_code, t.invoice_date
    ) AS transaction_key,
    c.customer_key,
    p.product_key,
    d.date_key,
    t.invoice_id,
    t.is_cancelled,
    t.quantity,
    t.unit_price,
    t.total_cost
    FROM {{ref('int_online_retail')}} t
    INNER JOIN {{ ref('dim_customer') }} c ON c.customer_id = t.customer_id
    INNER JOIN {{ ref('dim_product') }} p ON p.stock_code = t.stock_code
    INNER JOIN {{ ref('dim_date') }} d ON d.invoice_date = t.invoice_date

I run the following command to build the dimensions and the fact table:

dbt run --select dim_customer dim_product dim_date fact_transaction

Validate star schema

Having implemented the star schema, I add the following validation tests:

• Test the uniqueness and integrity of the surrogate keys.
• Test the foreign key relationships.

The final schema file:

version: 2

models:

  - name: dim_customer
    columns:
      - name: customer_key
        tests:
          - not_null
          - unique

      - name: customer_id

      - name: country


  - name: dim_product
    columns:
      - name: product_key
        tests:
          - not_null
          - unique

      - name: stock_code

      - name: description


  - name: dim_date
    columns:
      - name: date_key
        tests:
          - not_null
          - unique

      - name: invoice_date

      - name: i_year

      - name: i_month

      - name: i_day


  - name: fact_transaction
    columns:
      - name: transaction_key
        tests:
          - not_null
          - unique

      - name: customer_key
        tests:
          - not_null
          - relationships:
              to: ref('dim_customer')
              field: customer_key

      - name: product_key
        tests:
          - not_null
          - relationships:
              to: ref('dim_product')
              field: product_key

      - name: date_key
        tests:
          - not_null
          - relationships:
              to: ref('dim_date')
              field: date_key

      - name: invoice_id

      - name: is_cancelled

      - name: quantity

      - name: unit_price

      - name: total_cost

I check the validation by running:

dbt test --select dim_customer dim_product dim_date fact_transaction

Figure 4. Snowsight showcases the successful creation of the dimensions and the fact table, along with the population of fact_transaction

Finally, I drop invoices_raw_data from the tables as it is no longer required after the transformation process.

References:

• Python ingestion - GitHub repository
• Data transformation with dbt - GitHub repository
• Online Retail dataset - by Daqing Chen

Trademarks

• Python is a registered trademark of the Python Software Foundation.
• Snowflake is a registered trademark of Snowflake Inc.
• dbt and dbt-core are registered trademarks of dbt Labs, Inc.