DEV Community: kitchen_code

Building an Online Retail Sales & Cancellation Analysis Report With Power BI

kitchen_code — Thu, 27 Aug 2026 08:34:37 +0000

This article targets junior data engineers and data analysts who are familiar with:

ELT pipelines
Data warehouses
Star schemas
Snowflake

This article demonstrates basic data analysis and visualization in Power BI using the Online Retail data warehouse.

Note: This analysis uses the Online Retail dataset by Daqing Chen, made available through the UCI Machine Learning Repository under the CC BY 4.0 license.

Power BI: Online Retail Sales & Cancellation Analysis

1. Connect Power BI to Snowflake

I start by creating a Power BI role in Snowflake using the following statements.

CREATE ROLE POWERBI;

GRANT USAGE
ON WAREHOUSE ONLINE_RETAIL_WAREHOUSE
TO ROLE POWERBI;

GRANT USAGE
ON DATABASE ONLINE_RETAIL_DB
TO ROLE POWERBI;

GRANT USAGE
ON SCHEMA ONLINE_RETAIL_DB.RETAIL
TO ROLE POWERBI;

GRANT SELECT
ON ALL TABLES IN SCHEMA ONLINE_RETAIL_DB.RETAIL
TO ROLE POWERBI;

These queries create POWERBI, which is a new Power BI role. Then I grant the role privileges on the online_retail_warehouse and online_retail_db objects.

In Power BI, I select Snowflake as the data source option, as shown in Figure 1.

Figure 1. Select Snowflake as the data source in Power BI.

Next, I fill the required inputs to establish the connection with Snowflake. Figure 2 is an example of this step.

Figure 2. Submit Snowflake credentials to establish connection.

The required inputs are:

A server value, which is found in the account information section in Snowsight.
A warehouse identifier, which in this case is online_retail_warehouse.
An optional role, which I created in the previous SQL statements.

Once the connection is established, I select the fact table and dimension tables, and choose the load option. Figure 3 is a representation of this step.

Figure 3. Select the Online Retail fact table and dimension tables.

Next I choose the load method as shown in Figure 4.

Figure 4. Select Import method.

I choose Import because the data is relatively small and static. The Power BI Import option creates a copy of the data in its in-memory VertiPaq storage engine to ensure faster performance.

Figure 5 shows the imported tables in the Power BI data pane, confirming that the data has successfully been loaded.

Figure 5. Imported Online Retail tables in Power BI.

2. Verify the Star Schema

Figure 6 shows the star schema inside the Power BI model section.

Figure 6. Online Retail star schema on Power BI.

I verify that fact_transaction and its relationships with the dimension tables are correctly defined.

3. Define the business questions

To guide the analysis, I organize the business questions in four categories: customer analysis, time-based cancellation analysis, product cancellation analysis, geographic cancellation analysis.

Customer Analysis

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?

Time-based Cancellation Analysis

Which periods experience the highest number of cancellations?
How does the cancellation rate vary throughout the year?

Product Cancellation Analysis

Which products are most frequently associated with cancellations?

Geographic Cancellation Analysis

Which countries have the highest cancellation rate, and which have the lowest?

4. Define the analytical measures

I define the following analytical measures required to answer the business questions:

Total transactions: total number of transactions.
Cancelled transactions: number of cancelled transactions.
Cancellation rate: percentage of transactions that were cancelled.
Total revenue: total revenue generated.
Total quantity: total quantity of products purchased.
Average unit price: average price per unit.

With the measures identified, I can now implement them in Power BI using Data Analysis Expressions as follows.

In order to create a new measure, I right-click the fact table in the model view on Power BI, and select New measure.

Total transactions

Total Transaction = COUNTROWS(FACT_TRANSACTION)

I use the COUNTROWS aggregation function to count all the rows in fact_transaction and return the total number of transactions.

Cancelled transactions

Cancelled Transactions =
CALCULATE(
    COUNTROWS(FACT_TRANSACTION),
    FACT_TRANSACTION[is_cancelled] = True
)

I use the CALCULATE function to evaluate the COUNTROWS function under a modified filter context. The filter checks the value of the is_cancelled column, and only counts rows where its value is True.

Cancellation rate

Cancellation Rate = DIVIDE([Cancelled transactions], [Total Transaction])

To calculate the cancellation rate, I use the DIVIDE function to divide the Cancelled transactions measure by the Total transaction measure.

Total revenue

Total Revenue = SUM('FACT_TRANSACTION'[total_cost])

I use the SUM aggregation function to calculate the sum of the values in the total_cost column.

Total quantity

Total Quantity = SUM('FACT_TRANSACTION'[QUANTITY])

I use the SUM aggregation function to calculate the sum of the values in the quantity column.

Average unit price

Average Unit Price = AVERAGE('FACT_TRANSACTION'[UNIT_PRICE])

I use the AVERAGE aggregation function to calculate the arithmetic mean of the unit_price values.

5. Build the visualizations

5.1 Customer Analysis

Which customers generate the most cancelled transactions?

To answer this question, I create a clustered bar chart. I navigate to the Report view in Power BI. Figure 7 shows the resulting visualization.

Figure 7. Clustered bar chart representing the Top ten customers by cancelled transactions.

I select the Cancelled transactions measure for the X-axis. The Y-axis represents customer_id from dim_customer. I filter the chart to display the top ten customers with the highest number of cancelled transactions.

Which customers generate the most transactions while having the lowest cancellation rate?

I choose a scatter chart to represent the transaction volume vs cancellation rate by customer. Figure 8 shows the resulting visualization.

Figure 8. Scatter chart representing the transaction volume vs. cancellation rate by customer.

I select the Total transaction measure for the X-axis, the Cancellation Rate measure for the Y-axis, and customer_id from dim_customer for the value.

Customers with the highest total transaction volumes and the lowest cancellation rates are positioned toward the bottom-right.

Which customers generate the most revenue?

I add a clustered bar chart to visualize the customers who generate the most revenue. Figure 9 shows the resulting visualization.

Figure 9. Clustered bar chart representing the top ten customers by total revenue.

I select the Total Revenue measure for the X-axis, while the Y-axis represents customer_id from dim_customer. I filter the chart to display the top ten customers with the highest total revenue.

Which customers purchase the largest quantity of products?

I create a clustered bar chart to represent the customers who purchased the largest quantities of products. Figure 10 shows the resulting visualization.

Figure 10. Clustered bar chart representing the top ten customers by total quantities purchased.

I select the Total Quantity measure for the X-axis, and customer_id from dim_customer for the Y-axis.

Which customers purchase products with the highest average price per unit?

I add a clustered bar chart to visualize the customers who purchased the products with the highest average price per unit. Figure 11 shows the resulting visualization.

Figure 11. Clustered bar chart representing the top ten customers by average price per unit.

I select the Average Unit Price measure for the X-axis, and customer_id from dim_customer for the Y-axis.

Which customers generate high revenue while maintaining a low cancellation rate?

I select a scatter chart to represent the revenue vs the cancellation rate by customer. Figure 12 shows the resulting visualization.

Figure 12. Scatter chart representing revenue vs. cancellation rate by customer.

I select the Cancellation Rate measure for the X-axis, the Total Revenue measure for the Y-axis, and customer_id from dim_customer for the value.

Customers with high revenues and low cancellation rates appear toward the upper-left.

5.2 Time-based Cancellation Analysis

Which periods experience the highest number of cancellations?

To answer this time-based question, I use a clustered column chart as shown in Figure 13.

Figure 13. Clustered column chart showcasing cancelled transactions by month.

I choose the Cancelled transactions measure for the Y-axis, and i_month from dim_date for the X-axis.

How does the cancellation rate vary throughout the year?

I create a line chart to visualize the fluctuation of the cancellation rate throughout the year. Figure 14 shows the resulting visualization.

Figure 14. Line chart showcasing monthly variation in the cancellation rate.

I select Cancellation Rate for the Y-axis, and i_month from dim_date for the X-axis.

Observation:

When the number of cancelled transactions and the cancellation rate are compared, it becomes clear that a higher number of cancellations does not always translate into a higher cancellation rate.

For instance, November's cancellation rate is 1% compared to January's 2%, and there are 1,073 cancelled transactions compared to 700 in January.

The reason for this discrepancy is that the cancellation rate takes into consideration the total number of transactions that took place during each period.

5.3 Product Cancellation Analysis

Which products are most frequently associated with cancellations?

I select a treemap to visualize the products that are frequently associated with cancellations as shown in Figure 15.

Figure 15. Treemap visualization of top ten products ranked by number of cancelled transactions.

I select Cancelled transactions for values, and stock_code from dim_product for category.

5.4 Geographic Cancellation Analysis

Which countries have the highest cancellation rate, and which have the lowest?

I select the Map to highlight the cancellation rates by country. Figure 16 shows the resulting visualization.

Figure 16. Map showing the cancellation rates by country.

6. Final Report

Figure 17 shows the final and Power BI report, regrouping different statistics and visualizations developped throughout the analysis.

Figure 17. Complete Power BI report.

Conclusion

During this exercise, I practiced

Connecting Power BI and Snowflake.
Importing Snowflake data into Power BI.
Developing analytical measures with Data Analysis Expressions.
Creating graphics to examine data, including:
- Bar Charts- Column charts.
- Scatter plots.
- Line charts.
- TreeMaps
- Maps

This workshop demonstrated how to utilize Power BI as an analytical and visualization layer on top of a data warehouse.

References

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

kitchen_code — Sun, 16 Aug 2026 04:51:20 +0000

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 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 absolute 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. The result is a table containing 536,638 records.
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') }}
),

customer_countries AS (
    SELECT
        customer_id,
        country,
        COUNT(*) AS country_count
    FROM clean_data
    GROUP BY
        customer_id,
        country
),

dim_customer_nsk AS (
    SELECT
        customer_id,
        country
    FROM customer_countries
    QUALIFY ROW_NUMBER() OVER (
        PARTITION BY customer_id
        ORDER BY country_count DESC, country
    ) = 1
),

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

Since customers can be associated with multiple countries, I select the most frequently occurring country for each customer. This ensures customer_id is unique in dim_customer while preserving the dominant country association.

dim_product:

WITH clean_data AS (
    SELECT
        *
    FROM {{ ref('int_online_retail') }}
),

product_descriptions AS (
    SELECT
        stock_code,
        description,
        COUNT(*) AS description_count
    FROM clean_data
    WHERE stock_code IS NOT NULL
      AND description IS NOT NULL
    GROUP BY
        stock_code,
        description
),

dim_product_nsk AS (
    SELECT
        stock_code,
        description
    FROM product_descriptions
    QUALIFY ROW_NUMBER() OVER (
        PARTITION BY stock_code
        ORDER BY description_count DESC
    ) = 1
),

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

Since stock_code can be associated with multiple descriptions, I select only the frequently occurring description for each product. This ensures the uniqueness of stock_code in dim_product while preserving the most frequent description.

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
        tests:
          - not_null
          - unique

      - name: country


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

      - name: stock_code
        tests:
          - not_null
          - unique

      - name: description


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

      - name: invoice_date
        tests:
          - not_null
          - unique

      - 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

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

References:

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.

Orchestrate ETL Pipelines With Apache Airflow®

kitchen_code — Mon, 10 Aug 2026 02:28:48 +0000

This document demonstrates a simple extract, transform, and load (ETL) pipeline orchestration with Apache Airflow.

The document is intended for data engineers who have a basic understanding of the following:

ETL processes.
Python / pandas.
PostgreSQL®.
Windows Subsystem for Linux (WSL 2).
Ubuntu.
Transmission Control Protocol and Internet Protocol (TCP/IP).

The document focuses on orchestration and does not explain the implementation of the ETL processes.

Open-Meteo Weather API ETL pipeline orchestration

Problem Definition

In my previous case studies, I built data pipelines that required manual execution. The absence of orchestration introduces repetitive tasks, and limits scalability. As the system grows, and data processing requirements increase, manual execution becomes impractical, and leads to operational delays.

In this case study, I orchestrate the Open-Meteo Weather API ETL pipeline. API data are offered under Attribution 4.0 International (CC BY 4.0).

This forecast ETL pipeline monitors daily Weather variables for the Casablanca metropolitan area.

The data pipeline workflow is defined as follows:

The pipeline retrieves a selected subset of Weather variables from the API using predefined parameters.
The pipeline flattens the required nested objects from the JSON response into DataFrames. This process generates two DataFrames. The first one stores daily Weather data for the area at two-hour intervals. The second one tracks the daily measurement units and their insertion date.
The pipeline loads the resulting data into two distinct relational tables in a PostgreSQL database.

For the full implementation of the ETL processes, see the GitHub repository.

Problem Solution

Modern data engineering offers a wide range of orchestration tools for batch processing and stream processing.

Apache Airflow® (or simply Airflow) is an open-source orchestration platform. It enables users to write, schedule, and monitor data workflows. Airflow supports batch processing, which makes the platform suitable for orchestrating the Open-Meteo Weather API ETL pipeline.

Implementation

Execution Environment

Figure 1 provides an overview of the pipeline architecture and the Airflow orchestration implementation.

Figure 1. Overview of the Open-Meteo API ETL pipeline architecture and the implementation of the Airflow orchestration.

WSL 2 provides an Ubuntu environment for installing and running Airflow, limiting the risk of compatibility issues. The Python project also runs inside WSL 2, while the PostgreSQL database runs locally on the host Windows operating system.

Airflow Installation

In this case study, I installed Airflow in two environments:

A main Airflow installation in the Ubuntu WSL 2 environment.
An Airflow installation inside the Python application's virtual environment using pip.

The main Airflow installation is responsible for the Airflow environment and its associated configuration, including the metadata database, administrator credentials, and other Airflow settings.

The installation inside the Python application's virtual environment provides the Airflow packages required by the project and allows Airflow to run within the same Python environment as the application's dependencies.

This distinction matters later when running Airflow standalone.

For further information on installing Airflow on WSL 2, see the official documentation.

Airflow logic

Airflow manages the workflow execution, while the Python application executes the ETL processes.

In Airflow, a Directed Acyclic Graph (DAG) is a file that defines the structure of workflows. A DAG contains all the tasks and their dependencies, which determine the order in which Airflow orchestrates the pipeline.

In order to orchestrate this ETL pipeline with Airflow, we define the following steps:

Create a DAG file.
Define the tasks and dependencies.
Configure the TCP/IP connection between the Python application and the PostgreSQL database.
Verify the creation of the DAG using the Airflow standalone GUI.
Monitor the data loading in the PostgreSQL database.

1. Create the DAG file

The Airflow project uses the airflow/dags directory in the root of the Python project. The dags subfolder contains all the DAG files that Airflow can use. In this case, I define a single file.

The project structure is as follows:

open_meteo_airflow_pipeline/ │ ├── airflow/ │ └── dags/ ├── pipeline/ │ ├── extract.py │ ├── transform.py │ └── load.py ├── main.py ├── .env ├── requirements.txt └── README.md

Inside dags, I create the open_meteo_etl_dag.py DAG file.

2. Define the tasks and dependencies

The following code is defined inside open_meteo_etl_dag.py:

import sys
import os

import pandas as pd

sys.path.append(
    os.path.abspath(
        os.path.join(os.path.dirname(__file__), "../..")
    )
)

from datetime import datetime, timedelta

from airflow.sdk import DAG, task

from pipeline.extract import fetch_data
from pipeline.transform import transform_data
from pipeline.load import load_data

default_args = {
    "owner": "kitchen_code",
    "retries": 5,
    "retry_delay": timedelta(minutes=3),
    "execution_timeout": timedelta(minutes=15),
}

with DAG(
    dag_id="etl_dag",
    start_date=datetime(2026, 8, 3),
    schedule="@daily",
    default_args=default_args,
    catchup=False
) as dag:

    @task
    def extract_task():
        return fetch_data()

    @task
    def transform_task(extracted_data):
        units, latitude, longitude, hourly_data = extracted_data
        units_df, weather_df = transform_data(
            units,
            latitude,
            longitude,
            hourly_data
        )

        units_path = "/tmp/weather_units.csv"
        weather_path = "/tmp/weather_data.csv"

        units_df.to_csv(units_path, index=False)
        weather_df.to_csv(weather_path, index=False)

        return units_path, weather_path


    @task
    def load_task(paths):
        units_path, weather_path = paths
        units_df = pd.read_csv(units_path)
        weather_df = pd.read_csv(weather_path)

        load_data(units_df, weather_df)


    extracted_data = extract_task()
    transformed_data = transform_task(extracted_data)
    load_task(transformed_data)

First, I import the Airflow components to define the DAG and the tasks:

from airflow.sdk import DAG, task

The DAG constructor accepts a wide range of parameters. In this file, I define the following parameters:

dag_id: defines the unique identifier that Airflow uses to identify the DAG.
start_date: defines the date from which the DAG's scheduled runs should begin.
schedule: defines the frequency at which Airflow schedules the DAG. In this project, the DAG is configured to run once per day using @daily.
default_args: defines a Python dictionary containing default configuration parameters. In this project, it contains: owner: identifies the owner responsible for this DAG. retries: defines the number of times Airflow should retry a task after a failure. retry_delay: defines the amount of time Airflow should wait before retrying to run a task. execution_timeout: defines the maximum amount of time a task is allowed to run before Airflow stops it due to a timeout.
catchup: tells a DAG whether to run all missed time intervals that occurred between its start_date and the current time.

Once I define the DAG parameters, I create the tasks. In this context, we need three tasks: one for each ETL process.

I use the @task decorator to create the first task, extract_task, for the extraction process:

@task
def extract_task():
    return fetch_data()

I import the function fetch_data() from the extract.py module, and I return the result generated by the function from the task.

Using @task, I create the second task transform_task for the transformation process:

 @task
    def transform_task(extracted_data):
        units, latitude, longitude, hourly_data = extracted_data
        units_df, weather_df = transform_data(
            units,
            latitude,
            longitude,
            hourly_data
        )

        units_path = "/tmp/weather_units.csv"
        weather_path = "/tmp/weather_data.csv"

        units_df.to_csv(units_path, index=False)
        weather_df.to_csv(weather_path, index=False)

        return units_path, weather_path

The tasks are connected as follows:

extracted_data = extract_task()
transformed_data = transform_task(extracted_data)

Since the tasks were created using the @task decorator, Airflow automatically handles the communication between them using the XComs.

Unlike the traditional approach, where XComs are managed with xcom_push() and xcom_pull(), the TaskFlow API handles this process automatically through task inputs and outputs. This defines the dependencies between tasks.

The returned values from extract_task are made available to transform_task through the extracted_data parameter. The transform_task is defined as follows:

The task receives the values returned by fetch_data(): units, latitude, longitude, hourly_data.
The task imports the transform_data() function from the transform.py module and passes the returned values of fetch_data() to it. The end result is two DataFrames: units_df and weather_df.
The task converts the two DataFrames into CSV files. This approach allows the next task to access the transformed data without passing the entire DataFrames through XCom. At last, the task returns the path of each file.

Using @task, I create the last task load_task, for the loading process:

    @task
    def load_task(paths):
        units_path, weather_path = paths
        units_df = pd.read_csv(units_path)
        weather_df = pd.read_csv(weather_path)

        load_data(units_df, weather_df)

The tasks are connected as follows:

transformed_data = transform_task(extracted_data)
load_task(transformed_data)

The load_task takes paths as a parameter. This allows the task to read the CSV files using their paths, and load their contents back into pandas DataFrame.

The task calls the load_data() function from the load.py module. This function executes the following operations:

Establishes a connection to the PostgreSQL database using the psycopg driver.
Executes Data Definition Language statements to create the weather_data, and weather_units tables.
Executes Data Manipulation Language statements to insert the transformed data into the tables.

3. Configure the TCP/IP connection between the Python application and the PostgreSQL database.

The Python application runs inside WSL 2, while the PostgreSQL database runs locally on the Windows host operating system. Since the application and the database run in different environments, the connection between them is established by TCP/IP.

The psycopg driver requires several connection parameters. These parameters are stored in environment variables:

DB_NAME=
DB_USER=
DB_PASSWORD=
DB_HOST=
DB_PORT=

In my case, the main configuration issue concerns the DB_HOST parameter. Since the Python application runs inside WSL 2, the PostgreSQL server that runs on the host operating system cannot be reached using the default localhost address.

Therefore, I use the IP address of the Windows host as it is accessible from the WSL 2 environment.

To retrieve this IP address, run the following command in your Windows terminal:

ipconfig

Look for the IP address under the vEthernet (WSL (Hyper-V firewall)) network adapter. This is the IP address to use for the DB_HOST parameter.

Note: Other configuration issues can prevent the connection from being established. Verify that PostgreSQL is configured to accept connections from WSL 2. This includes its listen_addresses and authentication rules inside the pg_hba.conf file.

4. Verify the creation of the DAG using the Airflow standalone GUI.

The airflow standalone CLI command enables users to quickly spin up all core components of Airflow locally and simultaneously. This command also provides access to a web based graphic interface (GUI) that enables users to monitor the DAGs.

First, the admin credentials are required to access the GUI. To find the credentials: navigate inside the Airflow main installation package and look for either:

simple_auth_manager_passwords.json.generated

standalone_admin_password.txt

by running the following command:

cat simple_auth_manager_passwords.json.generated

This command returns the account credentials to access the web interface.

Figure 2 illustrates the commands used to run Airflow in standalone mode from WSL 2.

Figure 2. CLI commands to run Airflow in standalone mode from WSL 2.

I used the commands in Figure 2 to:

Navigate to the Python application.
Activate the Python virtual environment.
Check which Airflow installation is being used.
Run Airflow in standalone mode.

Once Airflow is running, navigate to localhost:8080. Airflow uses this port by default. Access to the web interface is granted after submitting user credentials (username and password).

Search for the DAG created for the pipeline by dag_id. In this case, the dag_id is etl_dag.

Figure 3 showcases the successful creation and registration of etl_dag.

Figure 3. Airflow web interface showcasing the successful creation and registration of the etl_dag.

Note: Make sure the DAG is activated by enabling the toggle.

5. Monitor the data loading in the PostgreSQL database.

To verify that the data is being loaded, access the PostgreSQL database using pgAdmin.

Figure 4 illustrates the insertion of the transformed data into the database using pgAdmin.

Figure 4. pgAdmin interface illustrating the successful insertion of the data into the weather_data table.

Conclusion

The orchestration was successfully implemented and verified. The DAG was created and registered, and its defined tasks were also validated. The transformed data was also successfully loaded into the PostgreSQL database.

It is important to note that the current implementation is not hosted. Therefore, Airflow must be manually started whenever the pipeline needs to be executed. To ensure continuous execution of the pipeline, it can be hosted on cloud infrastructure.

References

Trademark Notice

Python and the Python logos are trademarks of the Python Software Foundation (PSF).
Postgres, PostgreSQL and the Slonik Logo are trademarks or registered trademarks of the PostgreSQL Community Association of Canada, and used with their permission.
Apache Airflow, Airflow, and the Airflow logo are trademarks of the Apache Software Foundation.
Ubuntu is a registered trademark of Canonical Ltd.

Data Quality Management: Data Validation And Data Cleansing With Pandas and Pandera

kitchen_code — Sat, 01 Aug 2026 21:36:14 +0000

Data quality management (DQM) is a set of practices aimed at improving and maintaining the quality of an organisation's data. Effective DQM practices help ensure data accuracy, completeness, consistency, timeliness, uniqueness, and validity.

In this article, I explore how to leverage Python libraries to carry out two fundamental DQM procedures: data validation and data cleansing.

The dataset chosen for this exercise is the Kaggle dataset: Cafe Sales - Dirty Data for Cleaning Training by Ahmed Mohamed, under the CC BY-SA 4.0 license. It is a CSV file that comprises 10,000 cafe sales transactions across eight columns. Data quality issues are deliberately introduced, making it well suited for practicing DQM procedures.

To implement these procedures, Pandas is utilized for data cleansing, while Pandera is used for data validation.

Data validation requires establishing and enforcing business rules and checks, according to IBM's data validation page. Each organization defines its own rules and checks. Nevertheless, we'll focus on the most common checks:

Data type checks
Uniqueness checks
Format checks
Code checks
Consistency checks
Range checks

First, I created a data directory inside the Python application to store and access the CSV file. Next, I separated the extraction stage from the transformation stage by creating a dedicated module for each. Visit the GitHub repository to see the project structure.

The execution is applied in the main.py file:

from workflow.extract import extract_data
from workflow.transform import transform_data

def main():
    df = extract_data("data/dirty_cafe_sales.csv")
    transform_data(df)
main()

Extract the CSV file: extract.py

import pandas as pd

def extract_data(file_path):
    df = pd.read_csv(file_path)
    return df

DQM procedures - data validation and data cleansing: transform.py

I begin by creating a transform_data(df) function that accepts the DataFrame returned by extract_data in extract.py as a parameter.

Rename columns

    df.rename(columns={'Transaction ID': 'transaction_id', 'Item': 'item', 'Quantity': 'quantity', 'Price Per Unit': 'price_per_unit', 'Total Spent': 'total_spent', 'Payment Method': 'payment_method', 'Location': 'location', 'Transaction Date': 'transaction_date'}, inplace=True)

Drop duplicates

df = df.drop_duplicates()

This removes the rows that are 100% identical across all columns.

Extract dirty data

    def data_extraction():
        dirty_values = ["UNKNOWN", "ERROR"]
        dirty_rows = []
        rows_to_drop = []

        for index, row in df.iterrows():
            for value in row:
                if value in dirty_values or pd.isna(value):
                    dirty_rows.append(row)
                    rows_to_drop.append(index)
                    break

        dirty_dataframe = pd.DataFrame(dirty_rows)
        clean_dataframe = df.drop(index=rows_to_drop)

        clean_dataframe = clean_dataframe.reset_index(drop=True)

        return clean_dataframe, dirty_dataframe

    cd, dd = data_extraction()

The objective of data_extraction() is to separate clean rows from those containing data quality issues.

The function first defines dirty_values, which is a list of predefined invalid values (UNKNOWN, ERROR). It then iterates through each row of the DataFrame examining each cell.

If a cell value matches the predefined ones in dirty_values or is identified as a missing value (NaN), it is therefore classified as a "dirty row".

The dirty row is appended to the dirty_rows list, while its index in the original DataFrame df is stored in the rows_to_drop list.

Once all rows are processed, dirty_rows is then converted to the DataFrame dirty_dataframe. Figure 1 presents this DataFrame with all its records.

Figure 1. Visualization of dirty_dataframe and all its records containing data quality issues. This DataFrame stores 6911 rows extracted from the original DataFrame.

These dirty rows whose indexes are stored in rows_to_drop are removed from df, resulting in a new Dataframe: clean_dataframe. Finally, the index of clean_dataframe is reset to maintain sequential row numbering. Figure 2 presents the clean_dataframe.

Figure 2. Visualization of clean_dataframe and all its records containing no data quality issues. This DataFrame stores 3089 rows remaining from the original DataFrame.

In the next steps, we will focus only on clean_dataframe.

Data type conversion

The relevant columns are converted to their appropriate data types. This ensures that numeric values are represented as integers or floats, while dates are stored using Pandas' datetime type.

The data type conversion is implemented as follows:

    cd["quantity"] = cd["quantity"].astype(int)
    cd["price_per_unit"] = cd["price_per_unit"].astype(float)
    cd["total_spent"] = cd["total_spent"].astype(float)
    cd["transaction_date"] = pd.to_datetime(cd["transaction_date"], errors="coerce")

Data Type Check

To begin with, however, what is a data type check?

A data type check identifies values that violate the specified data type or do not conform to its expected length, precision, or scale. Therefore, after performing the data type conversion, Pandera is used to validate that the columns have the expected data types.

Pandera provides two approaches to data validation: DataFrame Model and DataFrame Schema.

For this exercise, I chose to use DataFrame Schema:

    data_type_check = pa.DataFrameSchema(
        {
            "transaction_id": pa.Column(str),
            "item": pa.Column(str),
            "quantity": pa.Column(int),
            "price_per_unit": pa.Column(float),
            "total_spent": pa.Column(float),
            "payment_method": pa.Column(str),
            "location": pa.Column(str),
            "transaction_date": pa.Column(DateTime)
        }
    )

    data_type_check.validate(cd)

After converting each relevant column to its appropriate data type, the data_type_check schema is used to validate that the DataFrame complies with its rules.

Note: Data validation can be performed either before or after data transformation, depending on the ETL pipeline design. In this case, the schema is applied after the type conversion to verify that the transformation was successful. Alternatively, it can be used before the data type conversion to identify records whose data types do not conform to the expected rules.

Uniqueness Check

A uniqueness check is applied to columns whose values must be unique.

In our case, the uniqueness check can logically be applied only to the transaction_id column. The other columns can store duplicate values.

    uniqueness_check = pa.DataFrameSchema(
        {
        "transaction_id": pa.Column(unique=True),
        }
    )

    uniqueness_check.validate(cd)

Format Check

A format check is applied to columns that require specific data formatting such as email addresses or phone numbers.

In our case, we conduct a format check on the transaction_id column to verify that all the values start with TXN_.

    format_check = pa.DataFrameSchema(
        {
            "transaction_id": pa.Column(
                str,
                checks=pa.Check(lambda s: s.str.startswith("TXN_"))
            )
        }
    )

    format_check.validate(cd)

Code Check

A code check determines whether a data value is valid by comparing it to a list of acceptable values.

In our case, we want to validate the payment_method column against a set of acceptable codes.

transaction_methods = [
        "Credit Card",
        "Digital Wallet",
        "Cash",
        "Other",
        "Cheque",
        "Bank Transfers"
    ]
    code_check = pa.DataFrameSchema(
        {
            "payment_method": pa.Column(
                str,
                checks=pa.Check.isin(transaction_methods)
            )
        }
    )

    code_check.validate(cd)

We create the transaction_methods list, which contains a set of valid payment methods. Then, we define the DataFrameSchema code_check that uses a validation check to verify that a value is in transaction_methods.

Consistency Check

Consistency checks are performed to verify that the relationships between two or more columns satisfy business rules.

In this exercise, we examine whether the total_spent values are consistent with the result of multiplying quantity by price_per_unit for each row.

    consistency_check = pa.DataFrameSchema(
        checks=[
            pa.Check(lambda df: (
                    df["total_spent"] == df["quantity"] * df["price_per_unit"]
                )
            )
        ]
    )

    consistency_check.validate(cd)

This schema uses a lambda function to define a custom validation rule.

Range Check

A range check is a data validation check that determines whether numerical data falls within a predefined range of minimum and maximum values.

    range_check = pa.DataFrameSchema(
        {
            "transaction_date": pa.Column(
                DateTime,
                checks=pa.Check(
                    lambda s: (
                        (s >= pd.Timestamp("2023-01-01")) &
                        (s < pd.Timestamp("2024-01-01"))
                    )
                )
            ),
            "quantity": pa.Column(
                int,
                checks=pa.Check.ge(0)
            ),
            "price_per_unit": pa.Column(
                float,
                checks=pa.Check.ge(0)
            ),
            "total_spent": pa.Column(
                float,
                checks=pa.Check.ge(0)
            )
        }
    )
    range_check.validate(cd)

We examine whether quantity, price_per_unit, and total_spent contain only non-negative values. We also verify that the dataset only includes rows with dates between January 1st, 2023, and December 31st, 2023.

Sort the DataFrame by transaction date in descending order

cd_sorted = cd.sort_values(by='transaction_date', ascending=False)

Figure 3 presents the result of this transformation.

Figure 3. Visualization of cd_sorted after conducting a sort by transaction_date in descending order. This DataFrame stores 3089 rows remaining from the original DataFrame.

Results

The results of these DQM procedures are as follows:

cd_sorted: a clean DataFrame containing no identified data quality issues.
dd: a dirty DataFrame containing the rows with data quality issues extracted from the original dataset. The objective was not to permanently discard these records. Instead, they are preserved in dd as they may contain valuable information that can be used for further investigation.

The two distinct DataFrames can then be loaded into the destination of choice.

References

Incremental Relational Normalization of Semi-Structured API Product Data: An ELT Pipeline Case Study

kitchen_code — Mon, 27 Jul 2026 01:50:15 +0000

Check out my second article. I would be really happy to have your feedback.

Incremental Relational Normalization of Semi-Structured API Product Data: An ELT Pipeline Case Study

kitchen_code — Sat, 25 Jul 2026 23:51:12 +0000

In my previous case study, I built an ETL pipeline using CSV files as the data source. Since CSV is a structured format and the dataset was relatively clean, the transformation stage was more predictable, allowing me to focus on understanding the overall ETL workflow rather than exploring each stage in great technical depth.

For this new case study, I wanted to explore a different type of challenge by building an ELT pipeline using semi-structured data. At the same time, I was exploring relational database design and normalization principles, and I realized that they would naturally fit into the transformation phase.

This article presents the development of an ELT pipeline that combines these objectives: ingesting nested JSON data from a REST API using a Python application, loading the raw data into a PostgreSQL database, and incrementally transforming it into a normalized relational data model through SQL-based transformations.

Methods

System architecture

Figure 1 illustrates the high-level architecture of the ELT pipeline implemented in this case study.

Figure 1. High-level architecture of the data ELT pipeline.

REST API Endpoint

The dataset was obtained from the public DummyJSON REST API, which is released under the MIT license. The API exposes multiple resources; however, but this exercise focuses on the /products endpoint.

To perform the extraction and loading stages of the pipeline, I developed a Python application within a virtual environment (venv). Following the Separation of Concerns (SoC) principle, I implemented each stage in a separate module to improve modularity.

Extraction Stage

First, I installed the requests library to enable the application to send HTTP requests to the DummyJSON REST API and retrieve JSON data from the /products endpoint.

The response was parsed into a Python dictionary, from which the products array was extracted. This allowed the application to iterate over each product individually before the loading stage.

Loading Stage

Using psycopg3, I established a connection between the application and the PostgreSQL database product_catalog.

Each product record was loaded as a raw JSONB document into the raw_products_data table, preserving the original structure before transformation. Figure 2 presents the data model of raw_products_data.

Figure 2. Data model of the raw_products_data table used to store raw product records as JSONB documents.

Figure 3 demonstrates that raw_products_data was successfully populated.

Figure 3. Raw product records stored as JSONB documents in the raw_products_data table before any SQL-based transformations or normalization.

The next step is to transform the raw data into a normalized relational data model.

Transformation Stage

This stage represents the main focus of this study. The transformation is centered on the application of normalization principles to convert the raw JSONB documents into a structured relational data model.

After establishing the initial relational model, higher normal forms are evaluated to determine whether further decomposition is required.

First Normal Form (1NF):

Since the JSONB documents contain nested arrays and objects, their attributes are not represented in atomic values. This violates the requirements of First Normal Form, which states that each attribute must contain an atomic value.

I created the products_stage table to store the atomic attributes associated with each product. Top-level scalar attributes were mapped directly to columns, while nested objects were flattened into individual columns. The tags, images, and reviews attributes were intentionally excluded because they represent multi-valued attributes or relationships requiring separate relational tables.

The following SQL statement creates the products_stage table:

CREATE TABLE IF NOT EXISTS products_stage (
  product_id INT PRIMARY KEY, 
  title TEXT, 
  description TEXT, 
  category TEXT, 
  price DECIMAL(10, 2), 
  discount_percentage DECIMAL(5, 2), 
  rating DECIMAL(5, 2), 
  stock INT, 
  brand VARCHAR(100), 
  sku TEXT, 
  weight DECIMAL(10, 2), 
  dimension_width DECIMAL(10, 2), 
  dimension_height DECIMAL(10, 2), 
  dimension_depth DECIMAL(10, 2), 
  warranty_information TEXT, 
  shipping_information TEXT, 
  availability_status VARCHAR(100), 
  return_policy TEXT, 
  minimum_order_quantity INT, 
  created_at TIMESTAMPTZ, 
  updated_at TIMESTAMPTZ, 
  barcode TEXT, 
  qr_code TEXT, 
  thumbnail TEXT
)

Compared with raw_products_data shown in Figure 2, Figure 4 illustrates the data model of the products_stage table.

Figure 4. Data model of the products_stage containing only atomic values.

After creating the table, the following INSERT statement was executed to populate it with data extracted from the JSONB documents.

INSERT INTO products_stage(
  product_id, title, description, category, 
  price, discount_percentage, rating, 
  stock, brand, sku, weight, dimension_width, 
  dimension_height, dimension_depth, 
  warranty_information, shipping_information, 
  availability_status, return_policy, 
  minimum_order_quantity, created_at, 
  updated_at, barcode, qr_code, thumbnail
) 
SELECT 
  (raw_data ->> 'id'):: INT, 
  raw_data ->> 'title', 
  raw_data ->> 'description', 
  raw_data ->> 'category', 
  (raw_data ->> 'price'):: DECIMAL(10, 2), 
  (
    raw_data ->> 'discountPercentage'
  ):: DECIMAL(5, 2), 
  (raw_data ->> 'rating'):: DECIMAL(5, 2), 
  (raw_data ->> 'stock'):: INT, 
  raw_data ->> 'brand', 
  raw_data ->> 'sku', 
  (raw_data ->> 'weight'):: DECIMAL(10, 2), 
  (
    raw_data -> 'dimensions' ->> 'width'
  ):: DECIMAL(10, 2), 
  (
    raw_data -> 'dimensions' ->> 'height'
  ):: DECIMAL(10, 2), 
  (
    raw_data -> 'dimensions' ->> 'depth'
  ):: DECIMAL(10, 2), 
  raw_data ->> 'warrantyInformation', 
  raw_data ->> 'shippingInformation', 
  raw_data ->> 'availabilityStatus', 
  raw_data ->> 'returnPolicy', 
  (
    raw_data ->> 'minimumOrderQuantity'
  ):: INT, 
  (raw_data -> 'meta' ->> 'createdAt'):: TIMESTAMPTZ, 
  (raw_data -> 'meta' ->> 'updatedAt'):: TIMESTAMPTZ, 
  raw_data -> 'meta' ->> 'barcode', 
  raw_data -> 'meta' ->> 'qrCode', 
  raw_data ->> 'thumbnail' 
FROM 
  raw_products_data

The INSERT statement uses PostgreSQL's JSONB operators (-> and ->>) to navigate the JSON structure, extract values, and cast them to the appropriate SQL data types.

Figure 5 presents a sample of the populated products_stage table in PostgreSQL, showing a subset of its columns.

Figure 5. Sample output of the populated products_stage table in PostgreSQL.

Although the products_stage table satisfies the requirements of the First Normal Form (1NF), the complete relational model is not yet finalized. The excluded attributes must still be extracted into dedicated tables to ensure that multi-valued attributes are properly represented withtin the relational model.

Create the product_images table

To normalize the multi-valued images attribute, I created the product_images table.

Since a single product can have multiple images, the relationship between products_stage and product_images is one-to-many (1:N). Each image is therefore stored as a separate row and associated with its corresponding product through a foreign key.

CREATE TABLE IF NOT EXISTS product_images(
    image_id SERIAL PRIMARY KEY,
    product_id INT,
    image_url TEXT,
    CONSTRAINT fk_product_images
        FOREIGN KEY(product_id)
        REFERENCES products_stage(product_id)
)

Populate the product_images table

INSERT INTO product_images(product_id, image_url)
SELECT (raw_data->>'id')::INT,
        jsonb_array_elements_text(raw_data->'images')
FROM raw_products_data

The jsonb_array_elements_text() function expands each element of the images array into a separate row, allowing each image URL to be stored as an atomic value.

The updated relational model is shown in Figure 6.

Figure 6. Updated relational model showing the one-to-many (1:N) relationship between products_stage and product_images.

Create the product_reviews table

The relationship between products_stage and the new product_reviews table is also a one-to-many relationship since one product can have multiple reviews.

The following SQL statement creates product_reviews:

CREATE TABLE IF NOT EXISTS product_reviews(
    review_id SERIAL PRIMARY KEY,
    product_id INT,
    review_rating DECIMAL(3,2),
    review_comment TEXT,
    review_date TIMESTAMPTZ,
    reviewer_name VARCHAR(100),
    reviewer_email TEXT,
    CONSTRAINT fk_product_reviews
        FOREIGN KEY(product_id)
        REFERENCES products_stage(product_id)
)

Populate product_reviews

INSERT INTO product_reviews (
    product_id,
    review_rating,
    review_comment,
    review_date,
    reviewer_name,
    reviewer_email
)
SELECT
    (raw_data->>'id')::INT,
    (review->>'rating')::DECIMAL(3,2),
    review->>'comment',
    (review->>'date')::TIMESTAMPTZ,
    review->>'reviewerName',
    review->>'reviewerEmail'
FROM raw_products_data,
     jsonb_array_elements(raw_data->'reviews') AS review;

The jsonb_array_elements() function was used to iterate over the reviews array of JSON objects in each row of raw_products_data. Each review object was expanded into a separate row, allowing its individual attributes to be extracted and stored as atomic values.

The updated relational schema is shown in Figure 7.

Figure 7. Updated relational model showing the one-to-many (1:N) relationship between the products_stage table and the product_reviews table.

Create the tags table

Unlike the images and reviews attributes, the relationship between products and tags is many-to-many (M:N). A product can have multiple tags, and a tag can be associated with many products.

To model this relationship, I created a tags table that stores each unique tag only once. The following SQL statement creates the tags table:

CREATE TABLE IF NOT EXISTS tags(
    tag_id SERIAL PRIMARY KEY,
    tag_name TEXT UNIQUE
)

Populate the tags table

INSERT INTO tags(tag_name)
SELECT DISTINCT
    jsonb_array_elements_text(raw_data->'tags')
FROM raw_products_data

I created the junction table product_tags to represent the many-to-many relationship between products with their corresponding tags through foreign keys.

Create the product_tags table

CREATE TABLE IF NOT EXISTS product_tags(
    product_id INT,
    tag_id INT, 
    PRIMARY KEY(product_id, tag_id),
    CONSTRAINT fk_product_tags_product
        FOREIGN KEY (product_id)
        REFERENCES products_stage(product_id),
    CONSTRAINT fk_product_tags_tag
        FOREIGN KEY (tag_id)
        REFERENCES tags(tag_id)
)

Populate the product_tags table

INSERT INTO product_tags (product_id, tag_id)
SELECT DISTINCT
    (raw_data->>'id')::INT,
    t.tag_id
FROM raw_products_data
CROSS JOIN jsonb_array_elements_text(raw_data->'tags') AS tag_name
INNER JOIN tags t
    ON t.tag_name = tag_name;

The CROSS JOIN with jsonb_array_elements_text() expands the tags array of each product into individual rows, producing one row for each "product-tag" combination. These tag values are then matched with the corresponding records in the tags table using an INNER JOIN, allowing the tag_id values to be retrieved. Finally, the DISTINCT keyword ensures that each (product_id, tag_id) pair is inserted only once into product_tags.

The updated relational schema is shown in Figure 8.

Figure 8. Updated relational model showing the many-to-many (M:N) relationship between products_stage and tags, implemented through the product_tags junction table.

Second Normal Form (2NF):

For a relational model to satisfy the requirements of the Second Normal Form (2NF), it must first comply with the rules of the First Normal Form (1NF).

In the previous section, all tables were normalized to satisfy this prerequisite.

The next step is to examine whether each table in the 1NF relational model satisfies the second requirement of 2NF. This involves determining whether a non-key attribute within a table depends on only a subset of a composite primary key than the entire key.

Analyze products_stage

products_stage complies with 2NF. Since the primary key of the table is composed of a single attribute product_id (see Figure 4), partial dependencies cannot occur. Furthermore, all non-key attributes describe the product associated with product_id and are therefore fully functionally dependent on the primary key.

Analyze product_images

product_images fulfills the conditions of 2NF. Since the primary key consists of a single attribute image_id (see product_images in Figure 6), partial dependencies cannot occur. In addition, the non-key attributes, product_id and image_url, describe the image identified by image_id and are therefore fully functionally dependent on the primary key.

A common misconception is to assume that image_url depends on product_id because a product can have multiple images. However, that is not how the table is designed.

Each row represents a single image entity identified by image_id, while product_id simply establishes the relationship between the image and its corresponding product.

Analyze product_reviews

product_reviews satisfies the criteria for 2NF. Since the primary key consists of a single attribute review_id (see product_reviews in Figure 7), partial dependencies cannot occur. Furthermore, all non-key attributes are fully functionally dependent on the primary key.

Each row represents a single review entity identified by review_id, while product_id simply establishes the relationship between the review and its corresponding product.

Analyze tags and product_tags

tags satisfies 2NF. Since the primary key consists of a single attribute tag_id (see tags in Figure 8), partial dependencies cannot occur. Furthermore, the non-key attribute tag_name is fully functionally dependent on the primary key.

Unlike the previous tables, the product_tags table uses a composite primary key composed of product_id and tag_id (see product_tags in Figure 8).

Each row represents the association between a product and a tag, uniquely identified by the combination of product_id and tag_id.

However, the table contains no non-key attributes so no partial dependencies can exist. As a result, product_tags also satisfies the requirements of 2NF.

Third Normal Form (3NF):

For a relational model to comply with the conditions of the Third Normal Form (3NF), it must first satisfy the rules of 2NF.

In the previous section, all tables were normalized to satisfy this prerequisite.

The next step is to determine whether each table in the 2NF relational model satisfies 3NF.

During the analysis of each table, potential functional dependency candidates are identified and evaluated to determine whether they could introduce transitive dependencies. In other words, the objective is to identify if any non-key attribute within a table is transitively dependent on another non-key attribute in the same table.

Analyze products_stage

During the analysis, it was observed that products with stock value lower than 8 were consistently assigned the value "Low stock" in their availability_status attribute. Conversely, products with a stock value of 8 or greater were consistently assigned "In stock". This suggests a possible dependency between stock and availability_status.

However, this observation alone is insufficient to establish the existence of a functional dependency. As a result, the 3NF status of this table could not be determined.

Although removing the availability_status attribute would eliminate the potential dependency, it would alter the source data without sufficient proof that the attribute is reundundant.

Since the objective in this case study is to normalize the source data rather than modify its underlying structure, the attribute availability_status was retained in the table.

Analyze product_images

Since a product may have multiple images, product_id does not functionally determine image_url; therefore, no transitive dependency exists between non-key attributes. Every non-key attribute depends directly on the primary key image_id, thus satisfying 3NF.

Analyze product_reviews

During the examination, it was determined that each reviewer_email value was consistently associated with one reviewer_name suggesting a possible functional dependency between the two non-key attributes. However, this observation alone is insufficient to determine the existence of a functional dependency. As a result, the 3NF status of this table could not be determined.

Analyze tags and product_tags

The tags table satisfies 3NF. It contains only one non-key attribute tag_name, which depends directly on the primary key tag_id. Therefore, no transitive dependency can exist. The same reasoning applies to the product_tags table.

Results

First Normal Form (1NF)

The raw product data was initially stored in raw_products_data as JSONB documents.

During the 1NF transformation, the semi-structured data was converted into a set of relational tables: products_stage, product_images, product_reviews, tags and product_tags.

Nested arrays and objects were eliminated and each attribute was stored as an atomic value while preserving the relationships between the entities through primary and foreign keys.

The resulting relational model satisfied the First Normal Form (1NF). Figure 9 presents the complete relational model obtained after the 1NF transformation.

Figure 9. Final relational schema satisfying the requirements of the First Normal Form 1NF, derived from the raw JSONB product data.

Second Normal Form (2NF)

The analysis showed that no partial dependencies exist within the relational model. Tables containing single-attribute primary keys (products_stage, product_images, product_reviews, and tags) cannot exhibit partial dependencies because every non-key attribute is fully functionally dependent on its primary key.

Although the product_tags table contains a composite primary key (product_id, tag_id), it stores no non-key attributes. Consequently, no partial dependencies can exist, and the table therefore does not violate the Second Normal Form (2NF) requirements.

No modifications were required ater the 2NF evaluation. The relational data model remains identical to the one shown in Figure 9.

Third Normal Form (3NF)

The 3NF analysis identified two categories of tables within the relational model.

The first category consists of the product_images, tags, and product_tags, for which no transitive dependencies were identified. These tables comply with 3NF.

The second category consists of products_stage and product_reviews, where potential functional dependencies were observed. However, the available data was insufficient to conclusively establish these dependencies. Therefore, their compliance with 3NF could not be conclusively established.

As a result no modifications to the relational model were required following the 3NF analysis. The final relational model therefore remains identical to the model presented in Figure 9.

The resulting relational model was normalized to 2NF. Although 3NF was established for some tables, the 3NF status of the complete relational model could not be determined.

Discussion

This case study set out to explore an ELT approach for handling semi-structured data and transforming it into a relational data model. Beyond this objective, the study also strengthened both my practical and theoretical skills.

The practical side involves performing HTTP requests with the requests library to consume a REST API in Python. It also involves working extensively with SQL in PostgreSQL by using JSONB operators and set-returning functions.

On the theorical side, the exercise provided hands-on application of normalization principles requiring the evaluation of the data models across the normal forms. In addition, it deepened my understanding of data modeling, particularly how relationships (1:N and M:N) are represented through foreign keys and junction tables as the model evolves.

References

Building my first ETL Pipeline: A Healthcare Management System case study

kitchen_code — Mon, 20 Jul 2026 00:26:02 +0000

Data engineering has long been a field of interest to me. Drawing upon my software engineering background, particularly in web development, I decided to undertake a structured learning path to master its core concepts. This article presents my first hands-on exercise in data engineering: the development of an ETL pipeline for a healthcare management system.

For this introductory case study, I chose to build an ETL pipeline because it provides a straightforward way to understand the three fundamental stages of the process: extraction, transformation and loading. By contrast, ELT pipelines are typically used within data warehouse or data lakehouse environments, adding another layer of concepts beyond the scope of this introductory exercise.

To support this case study, I selected the publicly available Healthcare Management System dataset published on Kaggle by Anouska Abhisikta. The dataset consists of five CSV files representing a healthcare management system: Appointment, Billing, Doctor, Medical Procedure and Patient. Its relational structure is particularly suitable for a introductory ETL pipeline practical demonstration. In addition, I deliberately selected a CSV-based dataset to avoid the additional complexity of formats such as JSON and Parquet files, or other data sources that would introduce distractions from the core ETL concepts explored in this exercise.

The objective is to build an ETL pipeline that extracts the data from these CSV files, applies data transformations, and loads the transformed data into a local PostgreSQL database.

METHOD:

The ETL pipeline was developed as a batch processing application using Python. A dedicated virtual environment (venv) was created in order to isolate project dependencies, with Pandas serving as the library for data manipulation. The source CSV files were stored in a data directory within the project and served as the input source throughout the pipeline.

The Python application follows the principle of separation of concerns, with each stage of the ETL pipeline being implemented in a separate module to improve readability, while main.py coordinates the execution of the pipeline (figure 1).

STAGE 1: EXTRACTION

During the extraction stage, each CSV file is independently read in its own dedicated module and the extraction function returns a Pandas DataFrame, marking the first stage of the ETL pipeline (Figure 2).

STAGE 2: TRANSFORMATION

The transformation stage demonstrates basic data processing techniques using the Python library Pandas. Given that this is a beginner case study, this data transformation is intentionally simple and is intended to understand the Transform phase rather than performing complex data manipulation strategies. Depending on the use case, different transformation rules may be applied to the same dataset.

Figure 3 illustrates an example of the data transformation performed on the Appointment dataset. This includes duplicate removal (drop_duplicates()), data type conversion, column renaming (rename()), and the creation of surrogate primary keys.

STAGE 3: LOAD

To load the transformed data, the Psycopg2 library was used to establish a connection between my Python application and a local PostgreSQL database. To centralize the database connection logic, a database.py file was created as shown in figure 4.

As an additional security layer, environment variables were used to avoid hardcoding credentials in the source code. Accordingly, the python-dotenv library was used to load these environment variables into the application at runtime.

Once the database connection is established, dedicated loading functions were implemented for each transformed dataset. These functions follow the same workflow. First, a database cursor is created to execute SQL statements. Then, the corresponding table is created if it does not already exist, and the transaction is committed before the insertion phase. This ensures that the table is successfully created and permanently saved, even if an error occurs during data insertion.

The transformed dataset is then iterated row by row so that each record is inserted into the corresponding table in the local PostgreSQL database. The transaction is committed again to permanently save the inserted data and the cursor is closed to release the associated resources. Figure 5 presents an example of the loading phase implementation.

OPTIONAL:

As an introduction to containerization concepts, the Python application was packaged using Docker. A Dockerfile was created to define the instructions required to build the Docker image as illustrated in figure 6.

Only the Python application was containerized, while the PostgreSQL database is kept running locally outside the image. Figure 7 presents the global architecture of this exercise.

RESULT

The main.py file plays a central role in coordinating the execution of the ETL pipeline. It acts as the entry point of the application by orchestrating the sequential execution of the different stages for each dataset entity.

The complete implementation of this file is shown in the code block below.

from extract.patient import extract_patient_data
from transform.patient import transform_patient_data
from load.patient import load_patient_data

from extract.doctor import extract_doctor_data
from transform.doctor import transform_doctor_data
from load.doctor import load_doctor_data

from extract.appointment import extract_appointment_data
from transform.appointment import transform_appointment_data
from load.appointment import load_appointment_data

from extract.billing import extract_billing_data
from transform.billing import transform_billing_data
from load.billing import load_billing_data

from extract.medical_proecdure import extract_med_procedure_data
from transform.medical_procedure import transform_med_procedure_data
from load.medical_procedure import load_med_procedure_data

from dotenv import load_dotenv
from database import get_connection

patient_data = extract_patient_data()
patient_table_data = transform_patient_data(patient_data)

doctor_data = extract_doctor_data()
doctor_table_data = transform_doctor_data(doctor_data)

appointment_data = extract_appointment_data()
appointment_table_data = transform_appointment_data(appointment_data)

billing_data = extract_billing_data()
billing_data_table = transform_billing_data(billing_data)

procedure_data = extract_med_procedure_data()
procedure_data_table = transform_med_procedure_data(procedure_data)

load_dotenv()
conn = get_connection()

if conn:
    try:
        print("Connected to database")
        load_patient_data(patient_table_data, conn)
        load_doctor_data(doctor_table_data, conn)
        load_appointment_data(appointment_table_data, conn)
        load_billing_data(billing_data_table, conn)
        load_med_procedure_data(procedure_data_table, conn)
    finally:
        conn.close()
else:
    print("Failed to connect to database")

After the execution of the main.py file, the pipeline successfully extracted data from the CSV files, applied the predefined transformations and loaded the transformed data into the PostgreSQL database. All relational tables were created as shown in Figure 8 and 9.

In conclusion, this study provided a practial experience with the fundamental concepts involved in developing a batch ETL pipeline. It enbales you to gain experience in:

Structuring Python projects with separation of concerns (modular packages per ETL stage).
Environment isolation with venv and dependency management with Pip.
Hands-on pandas transformations.
Connection Python applications to a local PostgreSQL database, with psycopg2 and safe credential handling using environment variables.
Basic Docker fundamentals (Dockerfile, building image, running containers).

As of now, the pipeline does not include an orchestration tool for scheduling or automating executions, and the PostgreSQL database is not containerized so the setup isn’t fully portable. These design choices are made to keep this exercise focused on understanding core ETL concepts while adding a basic application of Docker features. Future work will focus on different data sources such as Json, Parquet or perhaps a REST API and eventually explore ELT patterns.

Resources

Kaggle Healthcare Management System dataset (under the Apache 2.0 license)
Github repository of this exercise

DEV Community: kitchen_code

Building an Online Retail Sales & Cancellation Analysis Report With Power BI

Power BI: Online Retail Sales & Cancellation Analysis

1. Connect Power BI to Snowflake

2. Verify the Star Schema

3. Define the business questions

4. Define the analytical measures

5. Build the visualizations

5.1 Customer Analysis

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?

5.2 Time-based Cancellation Analysis

Which periods experience the highest number of cancellations?

How does the cancellation rate vary throughout the year?

5.3 Product Cancellation Analysis

Which products are most frequently associated with cancellations?

5.4 Geographic Cancellation Analysis

Which countries have the highest cancellation rate, and which have the lowest?

6. Final Report

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

Online Retail Data Warehouse with dbt and Snowflake

Dataset Analysis

Analytical requirements

Data quality requirements

Define the grain

Data modeling

Conceptual model

Logical model

Dimensions

Fact table: fact_transaction

Relationships

Physical model

Architecture overview

Data ingestion

Connect to Snowflake

dbt setup

Environment setup

Create project and connect to Snowflake

Data quality management

Staging

Data validation

Data type check

Uniqueness check

Data cleansing

Type cast and Column rename

Intermediate

Data validation

Range check

Format check

Data cleansing

Marts

Implement dimensions

Implement fact table

Validate star schema

Orchestrate ETL Pipelines With Apache Airflow®

Open-Meteo Weather API ETL pipeline orchestration

Problem Definition

Problem Solution

Implementation

Execution Environment

Airflow Installation

Airflow logic

Conclusion

Data Quality Management: Data Validation And Data Cleansing With Pandas and Pandera

Extract the CSV file: extract.py

DQM procedures - data validation and data cleansing: transform.py

Rename columns

Drop duplicates

Extract dirty data

Data type conversion

Data Type Check

Uniqueness Check

Format Check

Code Check

Consistency Check

Range Check

Fact table: `fact_transaction`