DEV Community: kitchen_code

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