Brazilian E-Commerce Data Pipeline

Introduction

30 Second Summary

Every time you place an online order, that transaction flows through data pipelines that validate, transform, and store it before anyone can analyze it. When those pipelines break or let bad data through, business teams make decisions based on incorrect numbers.

In this project, you will build a containerized data engineering pipeline that ingests real Brazilian e-commerce data, validates it with automated quality tests, and loads clean records into PostgreSQL and MinIO. You'll also create an interactive Streamlit dashboard where you can add new order items and immediately see analytics update.

What You'll Build

A production-style ETL pipeline using Docker Compose to orchestrate PostgreSQL, MinIO, pgAdmin, DuckDB, and Streamlit for end-to-end data processing and visualization.

By the end of this project, you'll have:

• A Docker Compose stack running PostgreSQL, MinIO, and pgAdmin for hybrid data storage and full database management.
• A pytest test suite using JSON Schema as an automated data quality gate that blocks bad records before they reach the database.
• An ETL pipeline that validates CSV data, loads it into PostgreSQL, and exports Parquet files to MinIO for fast analytics powered by DuckDB.
• A Streamlit dashboard with both analytics and a data entry form so new records flow into the pipeline and update metrics in real time.
• Secret Mission: A delivery performance analysis showing late vs on-time deliveries by customer state.

Are there any prerequisites?

This project assumes you have basic Python knowledge, familiarity with SQL, and Docker Desktop already installed on your machine.

Prior experience with an ETL pipeline (such as the weather ETL project) is helpful but not strictly required.

Before We Start

Before We Start...

Set Up Your Project and Infrastructure

A data pipeline needs somewhere to store raw data, somewhere to store processed data, and a way to manage it all. In this step, you'll spin up the infrastructure backbone using Docker containers.

You'll create the project folder structure, then define a Docker Compose stack with four services: PostgreSQL for your relational database, MinIO for S3-compatible object storage, pgAdmin for database management, and an init container that prepares MinIO with a storage bucket.

In this step, get ready to:

• Create the project directory structure.
• Write the docker-compose.yml to define all services.
• Start the containers and verify everything is healthy.

Create the project directory structure

Your pipeline needs separate folders to keep source code, tests, schemas, data files, and dashboards organized.

• In VS Code, open the integrated terminal by pressing Ctrl+` (macOS and Windows).
• Create the project subdirectories by running these commands:

mkdir src tests schemas data dashboards

• Confirm the folders exist by running:

ls

You should see src, tests, schemas, data, and dashboards listed in the output.

What are these folders for?

• src/ holds your ETL pipeline code.
• tests/ holds pytest data quality tests.
• schemas/ holds JSON Schema definitions for data validation.
• data/ holds the raw CSV dataset files from Kaggle.
• dashboards/ holds your Streamlit analytics app.

Create the Docker Compose file

Instead of running each container manually, Docker Compose lets you define your entire infrastructure stack in a single YAML file. One command brings everything up together.

• Create a new file called docker-compose.yml in the root of your project folder.
• Paste the following configuration into docker-compose.yml:

services:
  postgres:
    image: postgres:16-alpine
    container_name: olist-postgres
    environment:
      POSTGRES_USER: olist
      POSTGRES_PASSWORD: olist123
      POSTGRES_DB: ecommerce
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U olist -d ecommerce"]
      interval: 5s
      timeout: 5s
      retries: 5

  pgadmin:
    image: dpage/pgadmin4:latest
    container_name: olist-pgadmin
    environment:
      PGADMIN_DEFAULT_EMAIL: admin@olist.com
      PGADMIN_DEFAULT_PASSWORD: admin123
    ports:
      - "5050:80"
    volumes:
      - pgadmin_data:/var/lib/pgadmin
    depends_on:
      postgres:
        condition: service_healthy

  minio:
    image: minio/minio:latest
    container_name: olist-minio
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    ports:
      - "9000:9000"
      - "9001:9001"
    volumes:
      - minio_data:/data
    healthcheck:
      test: ["CMD", "mc", "ready", "local"]
      interval: 5s
      timeout: 5s
      retries: 5

  minio-init:
    image: minio/mc:latest
    container_name: olist-minio-init
    depends_on:
      minio:
        condition: service_healthy
    entrypoint: >
      /bin/sh -c "
      mc alias set minio http://minio:9000 minioadmin minioadmin;
      mc mb minio/ecommerce --ignore-existing;
      echo 'Bucket ecommerce created successfully';
      "

volumes:
  postgres_data:
  minio_data:
  pgadmin_data:

What does this configuration do?

• postgres runs PostgreSQL 16 with a database called ecommerce. The healthcheck uses pg_isready so other services know when the database is accepting connections.
• pgadmin provides a web-based UI for managing PostgreSQL. It waits for the database to be healthy before starting.
• minio provides S3-compatible object storage. Port 9000 handles the S3 API and port 9001 serves the web console.
• minio-init is a one-shot container that creates the ecommerce bucket in MinIO, then exits. It waits for MinIO to be healthy first.
• The volumes section at the bottom declares persistent storage so your data survives container restarts.

Why Docker Compose instead of running containers individually?

Running four containers manually means four separate terminal commands, each with environment variables, port mappings, and volume mounts. Docker Compose defines everything declaratively in one file. A single command brings the entire stack up or tears it down.

The depends_on with condition: service_healthy ensures services start in the right order. pgAdmin only starts after PostgreSQL is ready to accept queries. The init container only runs after MinIO is serving requests.

• Save docker-compose.yml.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

services:
  postgres:
    image: postgres:16-alpine
    container_name: olist-postgres
    environment:
      POSTGRES_USER: olist
      POSTGRES_PASSWORD: olist123
      POSTGRES_DB: ecommerce
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U olist -d ecommerce"]
      interval: 5s
      timeout: 5s
      retries: 5

  pgadmin:
    image: dpage/pgadmin4:latest
    container_name: olist-pgadmin
    environment:
      PGADMIN_DEFAULT_EMAIL: admin@olist.com
      PGADMIN_DEFAULT_PASSWORD: admin123
    ports:
      - "5050:80"
    volumes:
      - pgadmin_data:/var/lib/pgadmin
    depends_on:
      postgres:
        condition: service_healthy

  minio:
    image: minio/minio:latest
    container_name: olist-minio
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    ports:
      - "9000:9000"
      - "9001:9001"
    volumes:
      - minio_data:/data
    healthcheck:
      test: ["CMD", "mc", "ready", "local"]
      interval: 5s
      timeout: 5s
      retries: 5

  minio-init:
    image: minio/mc:latest
    container_name: olist-minio-init
    depends_on:
      minio:
        condition: service_healthy
    entrypoint: >
      /bin/sh -c "
      mc alias set minio http://minio:9000 minioadmin minioadmin;
      mc mb minio/ecommerce --ignore-existing;
      echo 'Bucket ecommerce created successfully';
      "

volumes:
  postgres_data:
  minio_data:
  pgadmin_data:

Seeing a YAML syntax error?

YAML is sensitive to indentation. Make sure every nested key is indented with exactly two spaces (not tabs). Check that the entrypoint block under minio-init uses the > character followed by the multiline string on the next line.

help me fix a YAML indentation error in my docker-compose.yml

Start the containers and verify services are healthy

With your Compose file saved, a single command pulls all images and starts every container in the correct order.

• Start all services in the background by running:

docker compose up -d

What does this command do?

The -d flag means "detached mode." Docker pulls any missing images, creates the containers, and starts them in the background. You get your terminal back immediately while the services continue running.

The first run takes 1-2 minutes as Docker downloads the PostgreSQL, MinIO, pgAdmin, and mc images. Subsequent runs are near-instant because images are cached locally.

• Check that all containers are running by running:

docker compose ps

You should see three containers with a status of "Up" or "healthy": olist-postgres, olist-minio, and olist-pgadmin. The olist-minio-init container will show "Exited (0)" because it ran once to create the bucket, then stopped.

• Verify the MinIO console by opening your browser and navigating to http://localhost:9001.
• Log in with username minioadmin and password minioadmin.
• You should see an ecommerce bucket listed in the MinIO console. This confirms the init container ran successfully.
• Verify pgAdmin by opening a new browser tab and navigating to http://localhost:5050.
• Log in with email admin@olist.com and password admin123.
• You should see the pgAdmin dashboard load. You'll connect it to PostgreSQL in a later step.

Containers not starting?

Make sure Docker Desktop is running. If you see "Cannot connect to the Docker daemon," Docker Desktop needs to be started first.

If a port conflict error appears (something like "port is already allocated"), another application is using port 5432, 9000, 9001, or 5050. Stop the conflicting application or change the host port in your docker-compose.yml.

help me troubleshoot my Docker Compose containers not starting

Your infrastructure is up and running. Next, you'll download the real e-commerce dataset and define the JSON schemas that will power your data quality checks.

Download the Dataset and Define JSON Schemas

Your infrastructure is running and ready to receive data. Now you need the actual source data that will flow through your pipeline.

In this step, you will download a real-world e-commerce dataset, define formal validation rules using JSON Schema, and set up your Python environment with all the libraries your pipeline needs.

In this step, get ready to:

• Download the Olist Brazilian E-Commerce dataset from Kaggle.
• Create JSON Schema files that define the expected structure of your data.
• Set up a Python virtual environment with all project dependencies.

Download the Olist dataset from Kaggle

The Olist dataset is a public Brazilian e-commerce dataset containing approximately 100,000 real orders from 2016 to 2018. It includes orders, products, payments, customers, and delivery information across 9 CSV files.

What is the Olist dataset?

Olist is a Brazilian e-commerce marketplace that published anonymized order data on Kaggle. The dataset contains real transactions spanning multiple product categories and Brazilian states.

For this project, you will use 3 of the 9 available CSV files. This keeps the pipeline focused while still representing realistic data volumes.

• Open the Olist dataset page on Kaggle: https://www.kaggle.com/datasets/olistbr/brazilian-ecommerce
• Sign in to your Kaggle account (or create a free one if you do not have one yet).
• Click the Download button near the top of the page to download the full ZIP file.
• Unzip the downloaded file.
• Copy these 3 CSV files into your data/ folder inside your project directory:

• olist_orders_dataset.csv
• olist_order_items_dataset.csv
• olist_order_payments_dataset.csv

• Confirm the files are in place by running:

ls data/

You should see olist_orders_dataset.csv, olist_order_items_dataset.csv, and olist_order_payments_dataset.csv listed in the output.

Don't see the CSV files?

Make sure you extracted the ZIP file before copying. The downloaded file from Kaggle is a .zip archive containing all the CSVs inside it.

Double-check that you copied the files into the data/ folder within your project directory, not into a different location.

Help me find my downloaded Kaggle files

Create the JSON Schema validation files

Before your pipeline loads any data into PostgreSQL, it needs to verify that each record has the correct structure. JSON Schema lets you define those rules in a portable, language-agnostic format.

What is JSON Schema?

JSON Schema is a standard for describing the expected structure of JSON data. You define which fields must exist, what types they should be, and any constraints like minimum values or allowed options.

Your pipeline will validate every record against these schemas before insertion. Any record that fails validation gets skipped, keeping your database clean.

• In VS Code, create a new file called orders_schema.json inside the schemas/ folder, and paste the following schema:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "order_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "customer_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "order_status": {
      "type": "string",
      "enum": ["delivered", "shipped", "canceled", "unavailable", "invoiced", "processing", "created", "approved"]
    },
    "order_purchase_timestamp": {"type": "string"},
    "order_approved_at": {"type": ["string", "null"]},
    "order_delivered_carrier_date": {"type": ["string", "null"]},
    "order_delivered_customer_date": {"type": ["string", "null"]},
    "order_estimated_delivery_date": {"type": "string"}
  },
  "required": ["order_id", "customer_id", "order_status", "order_purchase_timestamp", "order_estimated_delivery_date"]
}

What does this schema do?

• order_id and customer_id must be exactly 32-character strings (Olist uses UUID hex identifiers without hyphens).
• order_status is restricted to 8 allowed values using an enum constraint.
• Delivery date fields accept either a string or null because not all orders have been delivered yet.
• The required array ensures critical fields are never missing.

• Save schemas/orders_schema.json.

You will not see a visible change yet. The pytest tests in Step 3 will use this schema to validate every order record.

✔️ Awesome, I've got everything!

Great. Make sure you have saved schemas/orders_schema.json before moving on.

ⓧ I'd like to double check the full code

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "order_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "customer_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "order_status": {
      "type": "string",
      "enum": ["delivered", "shipped", "canceled", "unavailable", "invoiced", "processing", "created", "approved"]
    },
    "order_purchase_timestamp": {"type": "string"},
    "order_approved_at": {"type": ["string", "null"]},
    "order_delivered_carrier_date": {"type": ["string", "null"]},
    "order_delivered_customer_date": {"type": ["string", "null"]},
    "order_estimated_delivery_date": {"type": "string"}
  },
  "required": ["order_id", "customer_id", "order_status", "order_purchase_timestamp", "order_estimated_delivery_date"]
}

• Create a second file called order_items_schema.json inside the schemas/ folder, and paste this schema:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "order_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "order_item_id": {"type": "integer", "minimum": 1},
    "product_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "seller_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "shipping_limit_date": {"type": "string"},
    "price": {"type": "number", "minimum": 0},
    "freight_value": {"type": "number", "minimum": 0}
  },
  "required": ["order_id", "order_item_id", "product_id", "seller_id", "price", "freight_value"]
}

What does this schema do?

• order_item_id must be a positive integer (each item within an order gets a sequential number).
• price and freight_value must be numbers with a minimum of 0, preventing negative amounts from entering your database.
• ID fields (order_id, product_id, seller_id) are all validated as exactly 32-character strings.

• Save schemas/order_items_schema.json.

• Confirm both schema files exist by running:

ls schemas/

You should see orders_schema.json and order_items_schema.json in the output.

✔️ Awesome, I've got everything!

Great. Make sure both schema files are saved before moving on.

ⓧ I'd like to double check the full code

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "order_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "order_item_id": {"type": "integer", "minimum": 1},
    "product_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "seller_id": {"type": "string", "minLength": 32, "maxLength": 32},
    "shipping_limit_date": {"type": "string"},
    "price": {"type": "number", "minimum": 0},
    "freight_value": {"type": "number", "minimum": 0}
  },
  "required": ["order_id", "order_item_id", "product_id", "seller_id", "price", "freight_value"]
}

Install the Python dependencies

Your pipeline uses several Python libraries for database access, object storage, data validation, analytics, and testing. You will pin every version in a requirements.txt file and install them in an isolated virtual environment.

• Create a new file called requirements.txt in the root of your project folder, and paste the following:

psycopg2-binary==2.9.12
boto3==1.43.10
duckdb==1.5.2
streamlit==1.57.0
pyarrow==24.0.0
jsonschema==4.26.0
pytest==9.0.3
pandas==2.2.3

What are these packages?

• psycopg2-binary connects Python to PostgreSQL.
• boto3 talks to S3-compatible storage (MinIO in this project).
• duckdb is an in-process analytical database that queries Parquet files directly.
• streamlit builds the interactive analytics dashboard.
• pyarrow handles Parquet file reading and writing.
• jsonschema validates data records against the schemas you just created.
• pytest is the testing framework for your data quality gate.
• pandas provides DataFrame operations for data manipulation.

• Save requirements.txt.

✔️ Awesome, I've got everything!

Great. Make sure requirements.txt is saved in the root of your project folder (not inside src/ or schemas/).

ⓧ I'd like to double check the full code

psycopg2-binary==2.9.12
boto3==1.43.10
duckdb==1.5.2
streamlit==1.57.0
pyarrow==24.0.0
jsonschema==4.26.0
pytest==9.0.3
pandas==2.2.3

Now you need to verify your Python version and create a virtual environment. This project requires Python 3.11 or higher.

• Check your Python version by running:

python3 --version

✔️ I see version 3.11 or higher

Your Python version is ready. Continue to the next instruction below.

ⓧ I see an older version

Your system has a Python version below 3.11. On macOS, the default python3 often points to an older version even after installing a newer one.

• Install Python 3.11 or higher from https://www.python.org/downloads/.
• After installing, try running python3.11 --version or python3.12 --version to confirm the new version is available.
• Use the full version-specific command (e.g. python3.11) instead of python3 when creating your virtual environment below.

ⓧ Command not found

Python is not installed or not in your system path.

• Download Python 3.11 or higher from https://www.python.org/downloads/.
• Run the installer and make sure to check the option to add Python to your PATH.
• Close and reopen your terminal, then run python3 --version (macOS/Linux) or python --version (Windows) to confirm.

• Create a virtual environment by running:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms you are now working inside the isolated environment.

• Install all the project dependencies by running:

pip install -r requirements.txt

What does this command do?

pip install -r requirements.txt reads every line from your requirements file and installs that exact package version into your virtual environment. Pinning versions ensures everyone running this project gets identical behavior.

The installation takes a minute or two. When it finishes, you should see a success message listing the installed packages.

Seeing installation errors?

Make sure you activated your virtual environment first. You should see (venv) in your terminal prompt.

If you get a permissions error, confirm you are using pip from inside the virtual environment (not the system pip). Run which pip (macOS) or where pip (Windows) to check that it points to your venv/ folder.

Help me fix my pip install errors

Your data is downloaded, your validation rules are defined, and your Python environment is ready. Next up, you will write pytest tests that use these schemas as an automated quality gate before any data reaches PostgreSQL.

Build the Data Quality Tests with pytest

Your data is sitting in CSV files and your schemas define what valid records look like. But how do you know the data actually matches those schemas before loading it into PostgreSQL?

In this step, you'll write automated tests that act as a quality gate. These tests validate your CSV data against the JSON schemas you created, and check business rules like "prices must be positive." If bad data sneaks in, the tests catch it before it reaches your database.

In this step, get ready to:

• Write a pytest test file with schema validation functions.
• Add business rule tests for price and status values.
• Run the full test suite and verify all tests pass.

Write the test file with schema validation

Your test file needs helper functions to load schemas and CSV data, fixtures to share that data across tests, and test functions that validate records against the JSON schemas.

• Create a new file called test_data_quality.py inside your tests/ folder.
• Add the imports, path constants, and helper functions by pasting this code:

import json
import csv
from pathlib import Path

import pytest
from jsonschema import validate, ValidationError


DATA_DIR = Path(__file__).parent.parent / "data"
SCHEMA_DIR = Path(__file__).parent.parent / "schemas"


def load_schema(name: str) -> dict:
    schema_path = SCHEMA_DIR / f"{name}_schema.json"
    with open(schema_path) as f:
        return json.load(f)


def load_csv_rows(filename: str) -> list[dict]:
    filepath = DATA_DIR / filename
    with open(filepath, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        return list(reader)


@pytest.fixture
def orders_schema():
    return load_schema("orders")


@pytest.fixture
def order_items_schema():
    return load_schema("order_items")


@pytest.fixture
def orders_data():
    return load_csv_rows("olist_orders_dataset.csv")


@pytest.fixture
def order_items_data():
    return load_csv_rows("olist_order_items_dataset.csv")

What does this code do?

DATA_DIR and SCHEMA_DIR use Python's Path to resolve the data/ and schemas/ folders relative to the test file's location.

The load_schema() helper reads a JSON schema file by name. The load_csv_rows() helper reads a CSV file and returns a list of dictionaries (one per row).

The four @pytest.fixture functions provide test data to any test function that lists them as parameters. pytest automatically calls the fixture and injects the result.

You won't see a change yet. The next chunk adds the first test function that uses these fixtures.

Now add the first schema validation test. This function checks orders data against the orders JSON schema.

• Below the last fixture in tests/test_data_quality.py, paste this test function:

def test_orders_schema_valid(orders_data, orders_schema):
    """Validate first 100 order records against JSON schema."""
    errors = []
    for i, row in enumerate(orders_data[:100]):
        try:
            record = {
                "order_id": row["order_id"],
                "customer_id": row["customer_id"],
                "order_status": row["order_status"],
                "order_purchase_timestamp": row["order_purchase_timestamp"],
                "order_approved_at": row["order_approved_at"] or None,
                "order_delivered_carrier_date": row["order_delivered_carrier_date"] or None,
                "order_delivered_customer_date": row["order_delivered_customer_date"] or None,
                "order_estimated_delivery_date": row["order_estimated_delivery_date"],
            }
            validate(instance=record, schema=orders_schema)
        except ValidationError as e:
            errors.append(f"Row {i}: {e.message}")
    assert len(errors) == 0, f"Schema validation failed:\n" + "\n".join(errors[:5])

What does this test do?

It loops through the first 100 rows of the orders CSV and transforms each row into a dictionary matching the schema's expected shape. Empty strings for optional timestamp fields become None (matching the schema's ["string", "null"] type).

The validate() call from the jsonschema library checks each record against your orders_schema.json. If any record fails, it collects the error. The final assertion fails the test if any validation errors were found.

You won't see a change yet. The next chunk adds the matching test for order items.

Now add the second schema validation test for order items data.

• Below test_orders_schema_valid in tests/test_data_quality.py, paste this test function:

def test_order_items_schema_valid(order_items_data, order_items_schema):
    """Validate first 100 order item records against JSON schema."""
    errors = []
    for i, row in enumerate(order_items_data[:100]):
        try:
            record = {
                "order_id": row["order_id"],
                "order_item_id": int(row["order_item_id"]),
                "product_id": row["product_id"],
                "seller_id": row["seller_id"],
                "shipping_limit_date": row["shipping_limit_date"],
                "price": float(row["price"]),
                "freight_value": float(row["freight_value"]),
            }
            validate(instance=record, schema=order_items_schema)
        except ValidationError as e:
            errors.append(f"Row {i}: {e.message}")
    assert len(errors) == 0, f"Schema validation failed:\n" + "\n".join(errors[:5])

What does this test do?

This follows the same pattern as the orders test, but transforms CSV string values into the types the schema expects. The order_item_id is cast to int, while price and freight_value are cast to float.

CSV files store everything as strings. Without these type conversions, the schema's "type": "integer" and "type": "number" checks would always fail.

• Save tests/test_data_quality.py.

Seeing a syntax error?

Check that each function definition starts with def and is at the top level (no indentation). Also make sure you have a blank line between each function.

help me fix a syntax error in my pytest test file

Add tests for business rules

Schema validation catches structural issues like missing fields or wrong types. Business rule tests go further and check that values make sense in context.

• Below test_order_items_schema_valid in tests/test_data_quality.py, paste these two business rule tests:

def test_orders_no_empty_order_id(orders_data):
    """Business rule: order_id must never be empty."""
    empty_ids = [i for i, row in enumerate(orders_data) if not row["order_id"].strip()]
    assert len(empty_ids) == 0, f"Found {len(empty_ids)} rows with empty order_id"


def test_order_items_positive_price(order_items_data):
    """Business rule: price must be positive."""
    invalid = [
        i for i, row in enumerate(order_items_data)
        if float(row["price"]) < 0
    ]
    assert len(invalid) == 0, f"Found {len(invalid)} rows with negative price"

What do these tests check?

test_orders_no_empty_order_id scans every row in the orders CSV to ensure no order_id is blank or whitespace-only. An empty ID would break database inserts and joins.

test_order_items_positive_price checks that no item has a negative price. A negative price would corrupt revenue calculations downstream.

You won't see a change yet. The next chunk adds one more business rule test.

• Below test_order_items_positive_price, paste the final business rule test:

def test_orders_valid_status(orders_data):
    """Business rule: order_status must be in allowed values."""
    allowed = {"delivered", "shipped", "canceled", "unavailable", "invoiced", "processing", "created", "approved"}
    invalid = [
        (i, row["order_status"])
        for i, row in enumerate(orders_data)
        if row["order_status"] not in allowed
    ]
    assert len(invalid) == 0, f"Found {len(invalid)} rows with invalid status"

What does this test check?

This test defines the complete set of allowed order statuses and checks every row against it. If a new or misspelled status appears in the data, this test catches it before it reaches PostgreSQL.

The allowed set matches the enum values in your orders_schema.json. The schema validation test also catches this, but a dedicated business rule test makes failures easier to diagnose.

• Save tests/test_data_quality.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved the file.

ⓧ I'd like to double check the full code

import json
import csv
from pathlib import Path

import pytest
from jsonschema import validate, ValidationError


DATA_DIR = Path(__file__).parent.parent / "data"
SCHEMA_DIR = Path(__file__).parent.parent / "schemas"


def load_schema(name: str) -> dict:
    schema_path = SCHEMA_DIR / f"{name}_schema.json"
    with open(schema_path) as f:
        return json.load(f)


def load_csv_rows(filename: str) -> list[dict]:
    filepath = DATA_DIR / filename
    with open(filepath, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        return list(reader)


@pytest.fixture
def orders_schema():
    return load_schema("orders")


@pytest.fixture
def order_items_schema():
    return load_schema("order_items")


@pytest.fixture
def orders_data():
    return load_csv_rows("olist_orders_dataset.csv")


@pytest.fixture
def order_items_data():
    return load_csv_rows("olist_order_items_dataset.csv")


def test_orders_schema_valid(orders_data, orders_schema):
    """Validate first 100 order records against JSON schema."""
    errors = []
    for i, row in enumerate(orders_data[:100]):
        try:
            record = {
                "order_id": row["order_id"],
                "customer_id": row["customer_id"],
                "order_status": row["order_status"],
                "order_purchase_timestamp": row["order_purchase_timestamp"],
                "order_approved_at": row["order_approved_at"] or None,
                "order_delivered_carrier_date": row["order_delivered_carrier_date"] or None,
                "order_delivered_customer_date": row["order_delivered_customer_date"] or None,
                "order_estimated_delivery_date": row["order_estimated_delivery_date"],
            }
            validate(instance=record, schema=orders_schema)
        except ValidationError as e:
            errors.append(f"Row {i}: {e.message}")
    assert len(errors) == 0, f"Schema validation failed:\n" + "\n".join(errors[:5])


def test_order_items_schema_valid(order_items_data, order_items_schema):
    """Validate first 100 order item records against JSON schema."""
    errors = []
    for i, row in enumerate(order_items_data[:100]):
        try:
            record = {
                "order_id": row["order_id"],
                "order_item_id": int(row["order_item_id"]),
                "product_id": row["product_id"],
                "seller_id": row["seller_id"],
                "shipping_limit_date": row["shipping_limit_date"],
                "price": float(row["price"]),
                "freight_value": float(row["freight_value"]),
            }
            validate(instance=record, schema=order_items_schema)
        except ValidationError as e:
            errors.append(f"Row {i}: {e.message}")
    assert len(errors) == 0, f"Schema validation failed:\n" + "\n".join(errors[:5])


def test_orders_no_empty_order_id(orders_data):
    """Business rule: order_id must never be empty."""
    empty_ids = [i for i, row in enumerate(orders_data) if not row["order_id"].strip()]
    assert len(empty_ids) == 0, f"Found {len(empty_ids)} rows with empty order_id"


def test_order_items_positive_price(order_items_data):
    """Business rule: price must be positive."""
    invalid = [
        i for i, row in enumerate(order_items_data)
        if float(row["price"]) < 0
    ]
    assert len(invalid) == 0, f"Found {len(invalid)} rows with negative price"


def test_orders_valid_status(orders_data):
    """Business rule: order_status must be in allowed values."""
    allowed = {"delivered", "shipped", "canceled", "unavailable", "invoiced", "processing", "created", "approved"}
    invalid = [
        (i, row["order_status"])
        for i, row in enumerate(orders_data)
        if row["order_status"] not in allowed
    ]
    assert len(invalid) == 0, f"Found {len(invalid)} rows with invalid status"

Run the test suite and verify all tests pass

With all five test functions in place, it is time to run the quality gate and confirm your source data is clean.

• Run the test suite by running this command:

pytest tests/

• You should see output showing 5 tests collected and all passing, indicated by green dots and a line ending with 5 passed.

What does a passing test run look like?

pytest shows a dot (.) for each passing test. Five dots followed by 5 passed means every test function ran successfully. If any test fails, you see an F instead of a dot, with a detailed traceback showing which assertion failed and why.

Tests failing or not collected?

• Make sure your virtual environment is activated. If pytest is not found, run pip install pytest jsonschema again.
• Check that your CSV files are in the data/ folder with their exact Kaggle filenames (olist_orders_dataset.csv and olist_order_items_dataset.csv).
• Verify your JSON schema files are in the schemas/ folder and named orders_schema.json and order_items_schema.json.

help me debug failing pytest data quality tests

Why pytest plus JSON Schema instead of a heavier framework?

Frameworks like Great Expectations add orchestration, profiling, and a web UI. That power comes with setup complexity that slows down a focused pipeline.

pytest is already the standard Python testing framework, and jsonschema directly implements the JSON Schema specification. The schema files you write here are portable and language-agnostic. You could reuse them in a JavaScript front end, a Go microservice, or a CI linting step without rewriting any validation logic.

Your quality gate is in place. Any bad data that enters the pipeline will be caught before it reaches PostgreSQL. Next up, you'll build the ETL pipeline that uses these same validation functions to process the full dataset.

Build the ETL Pipeline

Your data quality tests proved the source data is valid. Now it is time to put those same validation patterns to work inside a real pipeline that moves data from CSV files into both PostgreSQL and MinIO.

The ETL script will extract rows from CSV, validate each record against JSON Schema (rejecting bad data), load clean records into PostgreSQL tables, and store columnar Parquet files in MinIO for fast analytics later.

In this step, get ready to:

• Write the ETL script that reads CSVs, validates records, and inserts into PostgreSQL.
• Add Parquet export logic that uploads analytical files to MinIO.
• Run the pipeline and verify loaded data in pgAdmin.

Write the ETL script with CSV reading and PostgreSQL loading

The ETL script lives in src/etl.py and reuses the same JSON Schema validation pattern from your tests. Records that fail validation are skipped instead of crashing the pipeline.

• Create a new file called etl.py inside your src/ folder.
• Add the imports, configuration constants, and data processing functions by pasting this code:

import csv
import json
import io
from pathlib import Path

import psycopg2
import boto3
import pyarrow as pa
import pyarrow.parquet as pq
from jsonschema import validate, ValidationError


DATA_DIR = Path(__file__).parent.parent / "data"
SCHEMA_DIR = Path(__file__).parent.parent / "schemas"

PG_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "ecommerce",
    "user": "olist",
    "password": "olist123",
}

MINIO_CONFIG = {
    "endpoint_url": "http://localhost:9000",
    "aws_access_key_id": "minioadmin",
    "aws_secret_access_key": "minioadmin",
}

BUCKET_NAME = "ecommerce"


def load_schema(name: str) -> dict:
    schema_path = SCHEMA_DIR / f"{name}_schema.json"
    with open(schema_path) as f:
        return json.load(f)


def load_csv(filename: str) -> list[dict]:
    filepath = DATA_DIR / filename
    with open(filepath, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        return list(reader)


def validate_records(records: list[dict], schema: dict, transform_fn) -> list[dict]:
    """Validate records against schema, return only valid ones."""
    valid = []
    invalid_count = 0
    for row in records:
        try:
            record = transform_fn(row)
            validate(instance=record, schema=schema)
            valid.append(record)
        except (ValidationError, ValueError, KeyError) as e:
            invalid_count += 1
    if invalid_count > 0:
        print(f"  Skipped {invalid_count} invalid records")
    return valid


def transform_order(row: dict) -> dict:
    return {
        "order_id": row["order_id"],
        "customer_id": row["customer_id"],
        "order_status": row["order_status"],
        "order_purchase_timestamp": row["order_purchase_timestamp"],
        "order_approved_at": row["order_approved_at"] or None,
        "order_delivered_carrier_date": row["order_delivered_carrier_date"] or None,
        "order_delivered_customer_date": row["order_delivered_customer_date"] or None,
        "order_estimated_delivery_date": row["order_estimated_delivery_date"],
    }


def transform_order_item(row: dict) -> dict:
    return {
        "order_id": row["order_id"],
        "order_item_id": int(row["order_item_id"]),
        "product_id": row["product_id"],
        "seller_id": row["seller_id"],
        "shipping_limit_date": row["shipping_limit_date"],
        "price": float(row["price"]),
        "freight_value": float(row["freight_value"]),
    }

What does this code do?

The top section defines connection details for PostgreSQL and MinIO so the rest of the script can reach both services.

• load_schema() reads JSON Schema files from the schemas/ folder. The same schemas your pytest tests use.
• load_csv() reads a CSV file into a list of dictionaries (one dict per row).
• validate_records() loops through every record, applies a transform function, validates against the schema, and collects only the valid ones. Invalid rows are counted and skipped.
• transform_order() and transform_order_item() convert raw CSV strings into properly typed Python dictionaries (casting integers and floats, converting empty strings to None).

• Save src/etl.py.
• Confirm the file exists by running:

ls src/

• You should see etl.py listed in the output.

Don't see etl.py?

Make sure you created the file inside the src/ folder, not in the project root. If src/ does not exist yet, create it first with mkdir src.

Help me create the etl.py file in the correct location.

• Now add the PostgreSQL table creation and data loading functions below the transform_order_item function in src/etl.py:

def create_tables(conn):
    """Create PostgreSQL tables for orders and order items."""
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS orders (
                order_id VARCHAR(32) PRIMARY KEY,
                customer_id VARCHAR(32) NOT NULL,
                order_status VARCHAR(20) NOT NULL,
                order_purchase_timestamp TIMESTAMP NOT NULL,
                order_approved_at TIMESTAMP,
                order_delivered_carrier_date TIMESTAMP,
                order_delivered_customer_date TIMESTAMP,
                order_estimated_delivery_date TIMESTAMP NOT NULL
            );
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS order_items (
                order_id VARCHAR(32) NOT NULL,
                order_item_id INTEGER NOT NULL,
                product_id VARCHAR(32) NOT NULL,
                seller_id VARCHAR(32) NOT NULL,
                shipping_limit_date TIMESTAMP NOT NULL,
                price NUMERIC(10,2) NOT NULL,
                freight_value NUMERIC(10,2) NOT NULL,
                PRIMARY KEY (order_id, order_item_id)
            );
        """)
    conn.commit()


def load_to_postgres(conn, table: str, records: list[dict]):
    """Insert records into PostgreSQL using batch insert."""
    if not records:
        return
    columns = list(records[0].keys())
    placeholders = ", ".join(["%s"] * len(columns))
    col_names = ", ".join(columns)
    query = f"INSERT INTO {table} ({col_names}) VALUES ({placeholders}) ON CONFLICT DO NOTHING"

    with conn.cursor() as cur:
        for record in records:
            values = [record[col] for col in columns]
            cur.execute(query, values)
    conn.commit()
    print(f"  Loaded {len(records)} records into {table}")

What does this code do?

• create_tables() runs SQL to create the orders and order_items tables with proper column types and constraints. CREATE TABLE IF NOT EXISTS means running the pipeline twice won't crash.
• load_to_postgres() builds a parameterized INSERT query from the record keys and inserts each row. ON CONFLICT DO NOTHING skips duplicates if the pipeline is re-run.

• Save src/etl.py.
• You will not see output yet. The next substep adds the MinIO upload function, and then you will run the full pipeline.

Add Parquet export and MinIO upload logic

Storing data as Parquet in object storage gives you a fast, columnar format that analytical engines like DuckDB can query directly without loading everything into memory first.

• Add the Parquet upload function below load_to_postgres in src/etl.py:

def upload_parquet_to_minio(records: list[dict], key: str):
    """Convert records to Parquet and upload to MinIO."""
    if not records:
        return
    table = pa.Table.from_pylist(records)
    buffer = io.BytesIO()
    pq.write_table(table, buffer)
    buffer.seek(0)

    s3 = boto3.client(
        "s3",
        endpoint_url=MINIO_CONFIG["endpoint_url"],
        aws_access_key_id=MINIO_CONFIG["aws_access_key_id"],
        aws_secret_access_key=MINIO_CONFIG["aws_secret_access_key"],
    )
    s3.put_object(Bucket=BUCKET_NAME, Key=key, Body=buffer.getvalue())
    print(f"  Uploaded {key} to MinIO ({len(records)} records)")

What does this code do?

• pa.Table.from_pylist(records) converts the list of dictionaries into a PyArrow table (the in-memory representation of columnar data).
• pq.write_table() serializes that table into the Parquet binary format in a memory buffer.
• boto3.client("s3") creates an S3-compatible client pointed at MinIO. MinIO speaks the same API as Amazon S3, so boto3 works without any changes.
• put_object() uploads the Parquet bytes into the ecommerce bucket under the given key path.

Why both PostgreSQL and Parquet?

PostgreSQL is great for row-level lookups and transactional writes (like the data entry form you will build in the next step). Parquet stored in object storage is optimized for column-scans across millions of rows, which is what analytical dashboards need.

Modern data teams often maintain both: an operational database for writes and a data lake for reads. This dual-storage pattern is exactly what your pipeline implements.

• Save src/etl.py.
• You will not see output yet. The next substep adds the orchestrator function that ties everything together.

Run the ETL pipeline and verify in pgAdmin

The run_pipeline() function orchestrates the full Extract, Validate, Load, and Store sequence in order.

• Add the pipeline orchestrator and entry point at the bottom of src/etl.py:

def run_pipeline():
    """Main ETL pipeline execution."""
    print("=" * 50)
    print("Brazilian E-Commerce ETL Pipeline")
    print("=" * 50)

    # Load schemas
    orders_schema = load_schema("orders")
    order_items_schema = load_schema("order_items")

    # Extract
    print("\n[1/5] Extracting data from CSV files...")
    raw_orders = load_csv("olist_orders_dataset.csv")
    raw_items = load_csv("olist_order_items_dataset.csv")
    print(f"  Orders: {len(raw_orders)} rows")
    print(f"  Order Items: {len(raw_items)} rows")

    # Validate
    print("\n[2/5] Validating data against JSON schemas...")
    valid_orders = validate_records(raw_orders, orders_schema, transform_order)
    valid_items = validate_records(raw_items, order_items_schema, transform_order_item)
    print(f"  Valid orders: {len(valid_orders)}")
    print(f"  Valid order items: {len(valid_items)}")

    # Load to PostgreSQL
    print("\n[3/5] Loading data into PostgreSQL...")
    conn = psycopg2.connect(**PG_CONFIG)
    create_tables(conn)
    load_to_postgres(conn, "orders", valid_orders)
    load_to_postgres(conn, "order_items", valid_items)
    conn.close()

    # Store as Parquet in MinIO
    print("\n[4/5] Storing Parquet files in MinIO...")
    upload_parquet_to_minio(valid_orders, "orders/orders.parquet")
    upload_parquet_to_minio(valid_items, "order_items/order_items.parquet")

    # Summary
    print("\n[5/5] Pipeline complete!")
    print(f"  PostgreSQL: orders ({len(valid_orders)}), order_items ({len(valid_items)})")
    print(f"  MinIO: orders/orders.parquet, order_items/order_items.parquet")
    print("=" * 50)


if __name__ == "__main__":
    run_pipeline()

What does this code do?

The run_pipeline() function executes the five stages in sequence:

• Loads both JSON schemas so validation knows what shape to expect.
• Extracts raw rows from the two CSV files.
• Validates every record against the matching schema, keeping only clean data.
• Connects to PostgreSQL, creates the tables, and inserts valid records.
• Converts valid records to Parquet and uploads both files to MinIO.

The if __name__ == "__main__" guard means the pipeline only runs when you execute the file directly, not when you import it.

• Save src/etl.py.
• Run the pipeline by executing this command in your terminal:

python src/etl.py

• You should see a step-by-step progress log ending with [5/5] Pipeline complete! and row counts for both tables and Parquet files.

Seeing a connection refused error?

• Make sure your Docker containers are still running. Run docker compose ps and check that olist-postgres and olist-minio show a healthy status.
• If the containers stopped, restart them with docker compose up -d and wait a few seconds before retrying.
• Verify your virtual environment is active. If you see a ModuleNotFoundError for psycopg2 or boto3, reactivate your virtual environment and re-run.

Help me debug my ETL pipeline connection error.

Now verify the data landed in PostgreSQL using pgAdmin.

• Open your browser and go to http://localhost:5050.
• Log in with email admin@olist.com and password admin123.
• Click Add New Server in the pgAdmin dashboard.
• In the General tab, enter olist-postgres as the server name.
• Switch to the Connection tab.
• Enter olist-postgres in the Host name/address field.
• Enter 5432 for Port.
• Enter ecommerce for Maintenance database.
• Enter olist for Username and olist123 for Password.
• Click Save.

You should now see the olist-postgres server in the left sidebar.

• Expand olist-postgres > Databases > ecommerce > Schemas > public > Tables.
• You should see the orders and order_items tables listed.
• Right-click the orders table and select View/Edit Data > First 100 Rows to see the loaded records.

What is pgAdmin?

pgAdmin 4 is the official web-based management tool for PostgreSQL. It lets you browse tables, run ad-hoc SQL queries, inspect schemas, and manage data without writing code. In a real data team, analysts and engineers use pgAdmin (or similar tools) to verify pipeline output and troubleshoot data issues.

Can't connect to the server in pgAdmin?

• The host must be olist-postgres (the Docker container name), not localhost. pgAdmin runs inside the Docker network and reaches PostgreSQL by container name.
• Double-check that the password is olist123 and the username is olist.

Help me connect pgAdmin to my PostgreSQL container.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/etl.py and that your pipeline output showed [5/5] Pipeline complete! with row counts.

ⓧ I'd like to double check the full code

Here is the complete src/etl.py file. Compare your code against this:

import csv
import json
import io
from pathlib import Path

import psycopg2
import boto3
import pyarrow as pa
import pyarrow.parquet as pq
from jsonschema import validate, ValidationError


DATA_DIR = Path(__file__).parent.parent / "data"
SCHEMA_DIR = Path(__file__).parent.parent / "schemas"

PG_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "ecommerce",
    "user": "olist",
    "password": "olist123",
}

MINIO_CONFIG = {
    "endpoint_url": "http://localhost:9000",
    "aws_access_key_id": "minioadmin",
    "aws_secret_access_key": "minioadmin",
}

BUCKET_NAME = "ecommerce"


def load_schema(name: str) -> dict:
    schema_path = SCHEMA_DIR / f"{name}_schema.json"
    with open(schema_path) as f:
        return json.load(f)


def load_csv(filename: str) -> list[dict]:
    filepath = DATA_DIR / filename
    with open(filepath, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        return list(reader)


def validate_records(records: list[dict], schema: dict, transform_fn) -> list[dict]:
    """Validate records against schema, return only valid ones."""
    valid = []
    invalid_count = 0
    for row in records:
        try:
            record = transform_fn(row)
            validate(instance=record, schema=schema)
            valid.append(record)
        except (ValidationError, ValueError, KeyError) as e:
            invalid_count += 1
    if invalid_count > 0:
        print(f"  Skipped {invalid_count} invalid records")
    return valid


def transform_order(row: dict) -> dict:
    return {
        "order_id": row["order_id"],
        "customer_id": row["customer_id"],
        "order_status": row["order_status"],
        "order_purchase_timestamp": row["order_purchase_timestamp"],
        "order_approved_at": row["order_approved_at"] or None,
        "order_delivered_carrier_date": row["order_delivered_carrier_date"] or None,
        "order_delivered_customer_date": row["order_delivered_customer_date"] or None,
        "order_estimated_delivery_date": row["order_estimated_delivery_date"],
    }


def transform_order_item(row: dict) -> dict:
    return {
        "order_id": row["order_id"],
        "order_item_id": int(row["order_item_id"]),
        "product_id": row["product_id"],
        "seller_id": row["seller_id"],
        "shipping_limit_date": row["shipping_limit_date"],
        "price": float(row["price"]),
        "freight_value": float(row["freight_value"]),
    }


def create_tables(conn):
    """Create PostgreSQL tables for orders and order items."""
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS orders (
                order_id VARCHAR(32) PRIMARY KEY,
                customer_id VARCHAR(32) NOT NULL,
                order_status VARCHAR(20) NOT NULL,
                order_purchase_timestamp TIMESTAMP NOT NULL,
                order_approved_at TIMESTAMP,
                order_delivered_carrier_date TIMESTAMP,
                order_delivered_customer_date TIMESTAMP,
                order_estimated_delivery_date TIMESTAMP NOT NULL
            );
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS order_items (
                order_id VARCHAR(32) NOT NULL,
                order_item_id INTEGER NOT NULL,
                product_id VARCHAR(32) NOT NULL,
                seller_id VARCHAR(32) NOT NULL,
                shipping_limit_date TIMESTAMP NOT NULL,
                price NUMERIC(10,2) NOT NULL,
                freight_value NUMERIC(10,2) NOT NULL,
                PRIMARY KEY (order_id, order_item_id)
            );
        """)
    conn.commit()


def load_to_postgres(conn, table: str, records: list[dict]):
    """Insert records into PostgreSQL using batch insert."""
    if not records:
        return
    columns = list(records[0].keys())
    placeholders = ", ".join(["%s"] * len(columns))
    col_names = ", ".join(columns)
    query = f"INSERT INTO {table} ({col_names}) VALUES ({placeholders}) ON CONFLICT DO NOTHING"

    with conn.cursor() as cur:
        for record in records:
            values = [record[col] for col in columns]
            cur.execute(query, values)
    conn.commit()
    print(f"  Loaded {len(records)} records into {table}")


def upload_parquet_to_minio(records: list[dict], key: str):
    """Convert records to Parquet and upload to MinIO."""
    if not records:
        return
    table = pa.Table.from_pylist(records)
    buffer = io.BytesIO()
    pq.write_table(table, buffer)
    buffer.seek(0)

    s3 = boto3.client(
        "s3",
        endpoint_url=MINIO_CONFIG["endpoint_url"],
        aws_access_key_id=MINIO_CONFIG["aws_access_key_id"],
        aws_secret_access_key=MINIO_CONFIG["aws_secret_access_key"],
    )
    s3.put_object(Bucket=BUCKET_NAME, Key=key, Body=buffer.getvalue())
    print(f"  Uploaded {key} to MinIO ({len(records)} records)")


def run_pipeline():
    """Main ETL pipeline execution."""
    print("=" * 50)
    print("Brazilian E-Commerce ETL Pipeline")
    print("=" * 50)

    # Load schemas
    orders_schema = load_schema("orders")
    order_items_schema = load_schema("order_items")

    # Extract
    print("\n[1/5] Extracting data from CSV files...")
    raw_orders = load_csv("olist_orders_dataset.csv")
    raw_items = load_csv("olist_order_items_dataset.csv")
    print(f"  Orders: {len(raw_orders)} rows")
    print(f"  Order Items: {len(raw_items)} rows")

    # Validate
    print("\n[2/5] Validating data against JSON schemas...")
    valid_orders = validate_records(raw_orders, orders_schema, transform_order)
    valid_items = validate_records(raw_items, order_items_schema, transform_order_item)
    print(f"  Valid orders: {len(valid_orders)}")
    print(f"  Valid order items: {len(valid_items)}")

    # Load to PostgreSQL
    print("\n[3/5] Loading data into PostgreSQL...")
    conn = psycopg2.connect(**PG_CONFIG)
    create_tables(conn)
    load_to_postgres(conn, "orders", valid_orders)
    load_to_postgres(conn, "order_items", valid_items)
    conn.close()

    # Store as Parquet in MinIO
    print("\n[4/5] Storing Parquet files in MinIO...")
    upload_parquet_to_minio(valid_orders, "orders/orders.parquet")
    upload_parquet_to_minio(valid_items, "order_items/order_items.parquet")

    # Summary
    print("\n[5/5] Pipeline complete!")
    print(f"  PostgreSQL: orders ({len(valid_orders)}), order_items ({len(valid_items)})")
    print(f"  MinIO: orders/orders.parquet, order_items/order_items.parquet")
    print("=" * 50)


if __name__ == "__main__":
    run_pipeline()

Your pipeline is now loading validated data into two storage layers. Next up, you will build a Streamlit dashboard that queries the Parquet files with DuckDB and lets users add new data entries that flow back through the pipeline.

Build the Streamlit Dashboard with Data Entry

Your ETL pipeline has loaded validated data into PostgreSQL and stored Parquet files in MinIO. Now it is time to make that data visible and interactive.

In this step you will build a Streamlit dashboard that reads Parquet directly from MinIO using DuckDB, displays key business metrics, and provides a data entry form so users can insert new order items that flow through the same quality gate.

In this step, get ready to:

• Create a Streamlit app that queries Parquet files in MinIO using DuckDB's httpfs extension.
• Build a data entry form that validates input against JSON Schema before inserting into PostgreSQL.
• Run the dashboard and verify analytics update when new data is added.

Write the DuckDB analytics dashboard

DuckDB can query Parquet files stored in S3-compatible object storage without loading them into memory first. You will connect DuckDB to MinIO and display three KPI metrics plus two visualizations.

• Create a new file called app.py inside your dashboards/ folder.
• Add the imports, configuration constants, page setup, and DuckDB connection by pasting this code:

import json
import io
from pathlib import Path
from datetime import datetime

import duckdb
import psycopg2
import boto3
import pyarrow as pa
import pyarrow.parquet as pq
import streamlit as st
from jsonschema import validate, ValidationError


SCHEMA_DIR = Path(__file__).parent.parent / "schemas"

PG_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "ecommerce",
    "user": "olist",
    "password": "olist123",
}

MINIO_CONFIG = {
    "endpoint_url": "http://localhost:9000",
    "aws_access_key_id": "minioadmin",
    "aws_secret_access_key": "minioadmin",
}

st.set_page_config(page_title="Olist E-Commerce Analytics", layout="wide")
st.title("\ud83c\udde7\ud83c\uddf7 Brazilian E-Commerce Analytics")
st.markdown("Powered by DuckDB + MinIO + Parquet | Managed by pgAdmin")


@st.cache_resource
def get_duckdb_connection():
    con = duckdb.connect()
    con.execute("INSTALL httpfs;")
    con.execute("LOAD httpfs;")
    con.execute("""
        CREATE SECRET (
            TYPE S3,
            KEY_ID 'minioadmin',
            SECRET 'minioadmin',
            ENDPOINT 'localhost:9000',
            USE_SSL false,
            URL_STYLE 'path'
        );
    """)
    return con


con = get_duckdb_connection()

What does this code do?

SCHEMA_DIR, PG_CONFIG, and MINIO_CONFIG define connection details for the schemas folder, PostgreSQL, and MinIO. These match the credentials in your docker-compose.yml.

The get_duckdb_connection() function installs and loads the httpfs extension so DuckDB can reach S3-compatible storage. The CREATE SECRET statement tells DuckDB how to authenticate with MinIO (endpoint, credentials, no SSL, path-style URLs).

The @st.cache_resource decorator ensures the connection is created once and reused across Streamlit reruns.

• Below the con = get_duckdb_connection() line, add the KPI metrics section:

st.header("\ud83d\udcca Key Metrics")
col1, col2, col3 = st.columns(3)

total_orders = con.execute(
    "SELECT COUNT(*) FROM read_parquet('s3://ecommerce/orders/orders.parquet')"
).fetchone()[0]

total_revenue = con.execute(
    "SELECT ROUND(SUM(price + freight_value), 2) FROM read_parquet('s3://ecommerce/order_items/order_items.parquet')"
).fetchone()[0]

avg_order_value = con.execute("""
    SELECT ROUND(AVG(order_total), 2) FROM (
        SELECT order_id, SUM(price + freight_value) as order_total
        FROM read_parquet('s3://ecommerce/order_items/order_items.parquet')
        GROUP BY order_id
    )
""").fetchone()[0]

col1.metric("Total Orders", f"{total_orders:,}")
col2.metric("Total Revenue", f"R$ {total_revenue:,.2f}")
col3.metric("Avg Order Value", f"R$ {avg_order_value:,.2f}")

What does this code do?

Each con.execute() call runs a SQL query against the Parquet files stored in MinIO. DuckDB reads them directly from the s3://ecommerce/ path without downloading them to disk first.

The three metrics show total orders, total revenue (price plus freight), and average order value. Streamlit's st.columns() lays them out side by side.

• Add the delivered orders over time chart below the KPI metrics:

st.header("\ud83d\udcc8 Delivered Orders Over Time")
revenue_by_month = con.execute("""
    SELECT
        strftime(order_purchase_timestamp::TIMESTAMP, '%Y-%m') as month,
        COUNT(*) as order_count
    FROM read_parquet('s3://ecommerce/orders/orders.parquet')
    WHERE order_status = 'delivered'
    GROUP BY month
    ORDER BY month
""").fetchdf()

st.line_chart(revenue_by_month.set_index("month")["order_count"])

What does this code do?

This query groups delivered orders by month using strftime to extract the year-month from each timestamp. The .fetchdf() method returns a pandas DataFrame that Streamlit renders as a line chart.

• Add the top sellers section below the chart:

st.header("\ud83c\udfc6 Top 10 Sellers by Revenue")
top_sellers = con.execute("""
    SELECT
        seller_id,
        COUNT(*) as items_sold,
        ROUND(SUM(price), 2) as total_revenue
    FROM read_parquet('s3://ecommerce/order_items/order_items.parquet')
    GROUP BY seller_id
    ORDER BY total_revenue DESC
    LIMIT 10
""").fetchdf()

st.dataframe(top_sellers, use_container_width=True)

What does this code do?

This query ranks sellers by total revenue from the order items Parquet file. The st.dataframe() widget displays the results as a sortable, interactive table.

• Save dashboards/app.py.

You will not see anything yet because Streamlit is not running. The next substep adds the data entry form, and then you will launch the app.

Seeing a red squiggly on an import?

If your editor shows import errors for duckdb or streamlit, make sure your Python virtual environment is selected as the interpreter in VS Code. All packages were installed in Step 2.

help me fix import errors in VS Code for my virtual environment

Add the data entry form

A dashboard that only reads data is useful, but a dashboard that also accepts new data simulates a real operational interface. You will add a form that validates input against your existing JSON Schema before inserting into PostgreSQL.

• In dashboards/app.py, add the form UI below the top sellers section:

st.markdown("---")
st.header("\u2795 Add New Order Item")
st.markdown("Insert a new order item into PostgreSQL. Analytics will update after re-exporting Parquet.")

with st.form("new_order_item_form"):
    col_a, col_b = st.columns(2)
    with col_a:
        new_order_id = st.text_input("Order ID (32 chars)", max_chars=32)
        new_product_id = st.text_input("Product ID (32 chars)", max_chars=32)
        new_price = st.number_input("Price (R$)", min_value=0.0, step=0.01)
    with col_b:
        new_order_item_id = st.number_input("Order Item ID", min_value=1, step=1)
        new_seller_id = st.text_input("Seller ID (32 chars)", max_chars=32)
        new_freight = st.number_input("Freight Value (R$)", min_value=0.0, step=0.01)

    new_shipping_limit = st.text_input(
        "Shipping Limit Date (YYYY-MM-DD HH:MM:SS)",
        value=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    )
    submitted = st.form_submit_button("Insert Order Item")

What does this code do?

The st.form() context manager groups all input widgets into a single submission. Nothing happens until the user clicks the submit button. This prevents Streamlit from re-running the app on every keystroke.

The form collects all fields required by your order_items_schema.json: order ID, item ID, product ID, seller ID, price, freight value, and shipping limit date.

• Below the form, add the submission handler and footer to complete the file:

if submitted:
    schema_path = SCHEMA_DIR / "order_items_schema.json"
    with open(schema_path) as f:
        item_schema = json.load(f)

    new_record = {
        "order_id": new_order_id,
        "order_item_id": int(new_order_item_id),
        "product_id": new_product_id,
        "seller_id": new_seller_id,
        "shipping_limit_date": new_shipping_limit,
        "price": float(new_price),
        "freight_value": float(new_freight),
    }

    try:
        validate(instance=new_record, schema=item_schema)

        conn = psycopg2.connect(**PG_CONFIG)
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO order_items (order_id, order_item_id, product_id, seller_id, shipping_limit_date, price, freight_value)
                   VALUES (%s, %s, %s, %s, %s, %s, %s)
                   ON CONFLICT DO NOTHING""",
                (new_order_id, int(new_order_item_id), new_product_id, new_seller_id, new_shipping_limit, new_price, new_freight),
            )
        conn.commit()
        conn.close()

        pg_conn = psycopg2.connect(**PG_CONFIG)
        with pg_conn.cursor() as cur:
            cur.execute("SELECT order_id, order_item_id, product_id, seller_id, shipping_limit_date::text, price::float, freight_value::float FROM order_items")
            rows = cur.fetchall()
            columns = ["order_id", "order_item_id", "product_id", "seller_id", "shipping_limit_date", "price", "freight_value"]
            records = [dict(zip(columns, row)) for row in rows]
        pg_conn.close()

        table = pa.Table.from_pylist(records)
        buffer = io.BytesIO()
        pq.write_table(table, buffer)
        buffer.seek(0)

        s3 = boto3.client("s3", endpoint_url=MINIO_CONFIG["endpoint_url"], aws_access_key_id=MINIO_CONFIG["aws_access_key_id"], aws_secret_access_key=MINIO_CONFIG["aws_secret_access_key"])
        s3.put_object(Bucket="ecommerce", Key="order_items/order_items.parquet", Body=buffer.getvalue())

        st.success(f"Inserted order item and updated analytics! ({len(records)} total items now)")
        st.rerun()

    except ValidationError as e:
        st.error(f"Validation failed: {e.message}")
    except Exception as e:
        st.error(f"Error: {e}")

st.markdown("---")
st.caption("Data source: Olist Brazilian E-Commerce Dataset | Pipeline: PostgreSQL + MinIO + DuckDB | Management: pgAdmin at localhost:5050")

What does this code do?

When the form is submitted, the handler first validates the new record against the same JSON Schema your pytest tests use. If validation passes, it inserts the record into PostgreSQL.

After the insert, it re-reads the entire order_items table from PostgreSQL, converts it to Parquet, and uploads it back to MinIO. This ensures the DuckDB analytics (which read from MinIO) reflect the new data immediately.

Finally, st.rerun() refreshes the page so the KPI metrics and charts update with the new record included.

Why re-export the entire Parquet file?

Parquet is an immutable columnar format. You cannot append a single row. The standard pattern is to re-export the full dataset after writes. For the ~100k records in this project, this takes under a second.

• Save dashboards/app.py.

Seeing a syntax error in the submission handler?

Check that the try: block is indented inside if submitted:, and that except ValidationError aligns with the try:. Python is sensitive to indentation levels.

help me fix a syntax or indentation error in my Streamlit form submission handler

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboards/app.py before continuing.

ⓧ I'd like to double check the full code

import json
import io
from pathlib import Path
from datetime import datetime

import duckdb
import psycopg2
import boto3
import pyarrow as pa
import pyarrow.parquet as pq
import streamlit as st
from jsonschema import validate, ValidationError


SCHEMA_DIR = Path(__file__).parent.parent / "schemas"

PG_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "ecommerce",
    "user": "olist",
    "password": "olist123",
}

MINIO_CONFIG = {
    "endpoint_url": "http://localhost:9000",
    "aws_access_key_id": "minioadmin",
    "aws_secret_access_key": "minioadmin",
}

st.set_page_config(page_title="Olist E-Commerce Analytics", layout="wide")
st.title("\ud83c\udde7\ud83c\uddf7 Brazilian E-Commerce Analytics")
st.markdown("Powered by DuckDB + MinIO + Parquet | Managed by pgAdmin")


@st.cache_resource
def get_duckdb_connection():
    con = duckdb.connect()
    con.execute("INSTALL httpfs;")
    con.execute("LOAD httpfs;")
    con.execute("""
        CREATE SECRET (
            TYPE S3,
            KEY_ID 'minioadmin',
            SECRET 'minioadmin',
            ENDPOINT 'localhost:9000',
            USE_SSL false,
            URL_STYLE 'path'
        );
    """)
    return con


con = get_duckdb_connection()

st.header("\ud83d\udcca Key Metrics")
col1, col2, col3 = st.columns(3)

total_orders = con.execute(
    "SELECT COUNT(*) FROM read_parquet('s3://ecommerce/orders/orders.parquet')"
).fetchone()[0]

total_revenue = con.execute(
    "SELECT ROUND(SUM(price + freight_value), 2) FROM read_parquet('s3://ecommerce/order_items/order_items.parquet')"
).fetchone()[0]

avg_order_value = con.execute("""
    SELECT ROUND(AVG(order_total), 2) FROM (
        SELECT order_id, SUM(price + freight_value) as order_total
        FROM read_parquet('s3://ecommerce/order_items/order_items.parquet')
        GROUP BY order_id
    )
""").fetchone()[0]

col1.metric("Total Orders", f"{total_orders:,}")
col2.metric("Total Revenue", f"R$ {total_revenue:,.2f}")
col3.metric("Avg Order Value", f"R$ {avg_order_value:,.2f}")

st.header("\ud83d\udcc8 Delivered Orders Over Time")
revenue_by_month = con.execute("""
    SELECT
        strftime(order_purchase_timestamp::TIMESTAMP, '%Y-%m') as month,
        COUNT(*) as order_count
    FROM read_parquet('s3://ecommerce/orders/orders.parquet')
    WHERE order_status = 'delivered'
    GROUP BY month
    ORDER BY month
""").fetchdf()

st.line_chart(revenue_by_month.set_index("month")["order_count"])

st.header("\ud83c\udfc6 Top 10 Sellers by Revenue")
top_sellers = con.execute("""
    SELECT
        seller_id,
        COUNT(*) as items_sold,
        ROUND(SUM(price), 2) as total_revenue
    FROM read_parquet('s3://ecommerce/order_items/order_items.parquet')
    GROUP BY seller_id
    ORDER BY total_revenue DESC
    LIMIT 10
""").fetchdf()

st.dataframe(top_sellers, use_container_width=True)

st.markdown("---")
st.header("\u2795 Add New Order Item")
st.markdown("Insert a new order item into PostgreSQL. Analytics will update after re-exporting Parquet.")

with st.form("new_order_item_form"):
    col_a, col_b = st.columns(2)
    with col_a:
        new_order_id = st.text_input("Order ID (32 chars)", max_chars=32)
        new_product_id = st.text_input("Product ID (32 chars)", max_chars=32)
        new_price = st.number_input("Price (R$)", min_value=0.0, step=0.01)
    with col_b:
        new_order_item_id = st.number_input("Order Item ID", min_value=1, step=1)
        new_seller_id = st.text_input("Seller ID (32 chars)", max_chars=32)
        new_freight = st.number_input("Freight Value (R$)", min_value=0.0, step=0.01)

    new_shipping_limit = st.text_input(
        "Shipping Limit Date (YYYY-MM-DD HH:MM:SS)",
        value=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    )
    submitted = st.form_submit_button("Insert Order Item")

if submitted:
    schema_path = SCHEMA_DIR / "order_items_schema.json"
    with open(schema_path) as f:
        item_schema = json.load(f)

    new_record = {
        "order_id": new_order_id,
        "order_item_id": int(new_order_item_id),
        "product_id": new_product_id,
        "seller_id": new_seller_id,
        "shipping_limit_date": new_shipping_limit,
        "price": float(new_price),
        "freight_value": float(new_freight),
    }

    try:
        validate(instance=new_record, schema=item_schema)

        conn = psycopg2.connect(**PG_CONFIG)
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO order_items (order_id, order_item_id, product_id, seller_id, shipping_limit_date, price, freight_value)
                   VALUES (%s, %s, %s, %s, %s, %s, %s)
                   ON CONFLICT DO NOTHING""",
                (new_order_id, int(new_order_item_id), new_product_id, new_seller_id, new_shipping_limit, new_price, new_freight),
            )
        conn.commit()
        conn.close()

        pg_conn = psycopg2.connect(**PG_CONFIG)
        with pg_conn.cursor() as cur:
            cur.execute("SELECT order_id, order_item_id, product_id, seller_id, shipping_limit_date::text, price::float, freight_value::float FROM order_items")
            rows = cur.fetchall()
            columns = ["order_id", "order_item_id", "product_id", "seller_id", "shipping_limit_date", "price", "freight_value"]
            records = [dict(zip(columns, row)) for row in rows]
        pg_conn.close()

        table = pa.Table.from_pylist(records)
        buffer = io.BytesIO()
        pq.write_table(table, buffer)
        buffer.seek(0)

        s3 = boto3.client("s3", endpoint_url=MINIO_CONFIG["endpoint_url"], aws_access_key_id=MINIO_CONFIG["aws_access_key_id"], aws_secret_access_key=MINIO_CONFIG["aws_secret_access_key"])
        s3.put_object(Bucket="ecommerce", Key="order_items/order_items.parquet", Body=buffer.getvalue())

        st.success(f"Inserted order item and updated analytics! ({len(records)} total items now)")
        st.rerun()

    except ValidationError as e:
        st.error(f"Validation failed: {e.message}")
    except Exception as e:
        st.error(f"Error: {e}")

st.markdown("---")
st.caption("Data source: Olist Brazilian E-Commerce Dataset | Pipeline: PostgreSQL + MinIO + DuckDB | Management: pgAdmin at localhost:5050")

Run the dashboard and test data entry

With the full file saved, you can now launch Streamlit and verify everything works end to end.

• Start the Streamlit app by running this command in your terminal:

streamlit run dashboards/app.py

Streamlit opens your browser at http://localhost:8501. You should see the title, three KPI metrics (Total Orders, Total Revenue, Avg Order Value), a line chart of delivered orders over time, and a top sellers table.

Dashboard shows an error about S3 or connection refused?

Make sure your Docker containers are still running. Run docker compose ps and confirm that olist-minio and olist-postgres show as healthy.

Also confirm you ran the ETL pipeline in Step 4. DuckDB needs the Parquet files to exist in MinIO before it can query them.

help me debug my Streamlit dashboard connection error to MinIO or PostgreSQL

• Scroll down to the Add New Order Item form.
• Enter a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4 in the Order ID field.
• Enter 1 in the Order Item ID field.
• Enter f1e2d3c4b5a6f1e2d3c4b5a6f1e2d3c4 in the Product ID field.
• Enter abcdef12345678abcdef12345678abcd in the Seller ID field.
• Enter 49.90 in the Price (R$) field.
• Enter 8.50 in the Freight Value (R$) field.
• Leave the Shipping Limit Date as the auto-filled current timestamp.
• Click Insert Order Item.

You should see a green success banner showing the total number of items now in the database. The page reruns automatically and the KPI metrics update to include your new record.

Why did the analytics update immediately?

When you clicked submit, the app inserted your record into PostgreSQL, re-exported the full order_items table as a fresh Parquet file to MinIO, then called st.rerun(). On the rerun, DuckDB reads the updated Parquet file and the metrics reflect the new data.

Form submission shows a validation error?

The JSON Schema requires all ID fields to be exactly 32 characters. Count the characters in your Order ID, Product ID, and Seller ID. Each must be exactly 32 characters long.

help me fix a JSON schema validation error when submitting the Streamlit form

Your pipeline is now fully interactive. Data flows from the form, through JSON Schema validation, into PostgreSQL, out to MinIO as Parquet, and back into DuckDB analytics. Next up, you will tackle the Secret Mission and extend this dashboard with delivery performance analysis.

Secret mission

Add a Delivery Performance Analysis

Your dashboard shows revenue and order metrics, but the business team wants to know: which states have the worst delivery performance? Extend your pipeline with customer location data and build a late delivery analysis that breaks down on-time vs late deliveries by Brazilian state.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Docker containers, so there are no ongoing cloud costs. However, the containers do consume memory and disk space while running.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Docker containers: olist-postgres, olist-minio, olist-pgadmin, and olist-minio-init.
• Docker volumes: postgres_data, minio_data, and pgadmin_data.
• Local project files: docker-compose.yml, requirements.txt, and all folders (src/, tests/, schemas/, data/, dashboards/).
• Python virtual environment used for running the ETL and dashboard.

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with the pipeline.

• Your Docker containers will keep running in the background.
• PostgreSQL data persists in the postgres_data volume.
• MinIO Parquet files persist in the minio_data volume.
• pgAdmin configuration persists in the pgadmin_data volume.
• You can access pgAdmin at localhost:5050 and the MinIO console at localhost:9001 anytime.

Pause - I'll come back to this later

Shut down running containers to free up memory. Your data volumes and project files stay intact so you can resume later.

• Open your terminal and navigate to the project folder containing docker-compose.yml.
• Stop all containers by running:

docker compose down

This stops and removes the containers but keeps your volumes intact. Next time, run docker compose up -d from the same folder to pick up where you left off.

Delete - I don't want to use this again

Remove all Docker resources and local files completely. This cannot be undone.

• Open your terminal and navigate to the project folder containing docker-compose.yml.
• Stop containers and remove all associated volumes by running:

docker compose down -v

• Navigate up to your Desktop and delete the project folder.

On macOS/Linux:

cd ~/Desktop
rm -rf olist-ecommerce

On Windows (PowerShell):

cd ~\Desktop
Remove-Item -Recurse -Force olist-ecommerce

• Deactivate your Python virtual environment if it is still active:

deactivate

The virtual environment files live inside the project folder, so deleting the folder removes them too.

Nice Work!

Nice Work!

You just built a production-grade data engineering pipeline from scratch. Real Brazilian e-commerce data flows through automated quality gates, into a relational database and object storage, and out through an interactive analytics dashboard.

You learned how to:

• Orchestrate a Docker Compose stack running PostgreSQL, MinIO, and pgAdmin for hybrid storage with a full management UI.
• Build JSON Schema-based data quality tests with pytest as an automated gate that blocks bad data before it reaches your database.
• Create an ETL pipeline that validates CSV data, loads it into PostgreSQL, exports Parquet files to MinIO, and serves real-time analytics through a Streamlit dashboard powered by DuckDB with an operational data entry form.
• Secret Mission: Add a delivery performance analysis showing late vs on-time deliveries by customer state with date range filtering.

Ready to quiz yourself?