Data Quality and Reporting System

Build an automated data quality pipeline with PostgreSQL and QuickSight.

Introduction

30 Second Summary

Every business runs on data. But when that data has missing fields, duplicates, or invalid values, decisions built on top of it silently fall apart.

In this project, you will build an automated data quality monitoring system that validates records in a PostgreSQL database, cleanses problematic data, and displays health metrics on a live dashboard.

What You'll Build

A Python-based data quality pipeline that uses Great Expectations to validate data in Amazon RDS PostgreSQL, automatically cleanses bad records, and surfaces quality KPIs through an Amazon QuickSight dashboard.

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

  • An Amazon RDS PostgreSQL instance loaded with sample operational data containing intentional quality issues like nulls, duplicates, and invalid formats.
  • A validation pipeline using Great Expectations that checks for nulls, duplicates, format violations, and value range errors.
  • An automated cleansing and scoring workflow that standardizes formats, removes duplicates, and logs quality scores to metrics tables in your database.
  • A live Amazon QuickSight dashboard displaying data quality KPIs including pass/fail rates, quality trends, and table-level health comparisons.
  • Secret Mission: Add threshold-based alerting that generates summary reports when quality scores drop below acceptable levels.

Are there any prerequisites?

You'll need an active AWS account, Python 3.10 installed on your machine, and VS Code ready to go. Some familiarity with SQL and PostgreSQL will help, but this project walks you through every step.

Set Up Your Environment and RDS Database

Before you can validate data quality, you need a database filled with real-world messiness. In this step, you will set up a Python project, spin up a cloud-hosted PostgreSQL database on Amazon RDS, and load it with intentionally dirty sample data.

The messy data includes nulls, duplicates, invalid email formats, and out-of-range values. These are exactly the kinds of problems a data quality pipeline is built to catch.

In this step, get ready to:
  • Install Python dependencies and configure the AWS CLI.
  • Create an Amazon RDS PostgreSQL instance using the Free Tier.
  • Load messy sample data into your database tables.
Install Python dependencies and configure AWS CLI

You need a dedicated project folder with an isolated Python environment. This keeps your data quality libraries separate from any other Python projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir data-quality-pipeline
cd data-quality-pipeline
  • Create and activate 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) at the start of your terminal prompt. This confirms the virtual environment is active.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Select the data-quality-pipeline folder on your Desktop, then click Open.
  • Create a new file called requirements.txt in the root of your project folder and paste the following contents:
great-expectations[postgresql]==1.18.1
psycopg2-binary==2.9.10
SQLAlchemy==2.0.41
python-dotenv==1.1.0

What are these packages?

  • great-expectations[postgresql] is the industry-standard data quality framework, installed with its PostgreSQL support.
  • psycopg2-binary is the PostgreSQL adapter that lets Python talk to your database.
  • SQLAlchemy is the SQL toolkit that Great Expectations uses under the hood.
  • python-dotenv loads environment variables from a .env file so you don't hardcode credentials.
  • Back in your terminal (make sure (venv) is still showing), install the dependencies by running:
pip install -r requirements.txt

This may take a minute as it downloads and installs the packages. You should see Successfully installed followed by a list of package names when it finishes.

Seeing an installation error?

Make sure your virtual environment is activated. You should see (venv) in your terminal prompt. If not, run the activate command again.

If you see a permissions error, ensure you are inside the virtual environment and not running with system Python.

help me fix a pip install error for great-expectations

Next, you need the AWS CLI configured so your Python scripts can connect to Amazon RDS later. The AWS CLI is a command-line tool for interacting with AWS services.

  • Check whether the AWS CLI is installed by running:
aws --version

✔️ I see a version number

The AWS CLI is ready. Move on to the configuration step below.

ⓧ Command not found

You need to install the AWS CLI. Follow the official installation guide for your operating system:

Install the AWS CLI

After installation, close and reopen your terminal, then run aws --version again to confirm it works.

  • Configure the AWS CLI with your credentials by running:
aws configure

What does aws configure ask for?

The command prompts you for four values. Enter them one at a time:

  • AWS Access Key ID and AWS Secret Access Key from your IAM user credentials.
  • Default region name (e.g. us-east-1). Choose the region closest to you.
  • Default output format. Press Enter to accept the default (json).
Create an Amazon RDS PostgreSQL instance

Your data quality pipeline needs a real database to validate against. You will create a PostgreSQL instance on Amazon RDS using the Free Tier template.

  • Open the AWS Management Console in your browser and sign in.
  • In the search bar at the top, type RDS and select RDS from the results.
  • In the left sidebar, click Databases.
  • Click Create database.
  • Under Choose a database creation method, select Standard create.
  • Under Engine options, select PostgreSQL.
  • For Engine Version, select PostgreSQL 16.
  • Under Templates, select Free tier.
  • Under Settings, enter dataqualitydb in the DB instance identifier field.
  • Set the Master username to postgres.
  • Enter a strong password in the Master password field. Write this down. You will need it shortly.
  • Under Instance configuration, confirm the instance class is db.t3.micro (Free Tier eligible).
  • Under Storage, confirm Allocated storage is 20 GB and storage type is gp2.
  • Under Connectivity, set Public access to Yes. This allows your local machine to connect.
  • Under VPC security group (firewall), select Create new. This creates a security group that allows your current IP address to access the database.
  • Expand Additional configuration. In the Initial database name field, enter dataqualitydb.
  • Scroll down and click Create database.

Cost warning

The Free Tier covers a db.t3.micro instance with 20 GB of gp2 storage. If your AWS account was created before July 15, 2025, you get 750 hours/month free for 12 months. Accounts created after that date receive a Free Plan with up to $200 in credits for 6 months.

Delete the RDS instance when you are done with this project to avoid charges after your free tier expires.

The instance takes 5-10 minutes to become available. Wait until the Status column shows Available before continuing.

  • Once available, click your database name to open its details.
  • On the Connectivity & security tab, find and copy the Endpoint value. It looks like dataqualitydb.xxxxxxxxxxxx.us-east-1.rds.amazonaws.com. Save this somewhere. You will use it in the next substep.

Now verify the security group allows your IP to connect on port 5432.

  • On the same Connectivity & security tab, click the active security group link under VPC security groups.
  • Click the Inbound rules tab.
  • Confirm there is a rule with Type set to PostgreSQL, Port range set to 5432, and Source set to your IP address.
  • If the rule does not exist, click Edit inbound rules, then Add rule. Set Type to PostgreSQL, and set Source to My IP. Click Save rules.

Why allow only your IP?

Security groups act as firewalls for your RDS instance. By restricting access to your IP address on port 5432, you prevent unauthorized connections. Only traffic from your machine can reach the database.

Load sample data into PostgreSQL

With the database running, you will connect from your local machine and load intentionally messy data. This data contains the kinds of quality issues your pipeline will detect and fix in the next steps.

  • In VS Code, create a new file called .env in the root of your data-quality-pipeline folder and paste the following:
DB_HOST=[[DB_HOST="your-rds-endpoint.region.rds.amazonaws.com"]]
DB_PORT=5432
DB_NAME=dataqualitydb
DB_USER=postgres
DB_PASSWORD=your-password-here
  • Replace the DB_HOST value with the endpoint you copied from the RDS console.
  • Replace your-password-here with the master password you set when creating the instance.
  • Save the file.

Keep .env out of version control

If you plan to use Git with this project, add .env to your .gitignore file. This prevents accidentally committing your database credentials.

Now you will create the setup script that builds your tables and inserts the sample data. This script is long because it defines five tables and two sets of deliberately messy records.

  • Create a new file called setup_database.py and paste the connection and table creation code:
import psycopg2
from dotenv import load_dotenv
import os

load_dotenv()

conn = psycopg2.connect(
    host=os.getenv("DB_HOST"),
    port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD")
)
conn.autocommit = True
cur = conn.cursor()

# Create tables
cur.execute("""
DROP TABLE IF EXISTS orders CASCADE;
DROP TABLE IF EXISTS customers CASCADE;
DROP TABLE IF EXISTS data_quality_metrics CASCADE;
DROP TABLE IF EXISTS quality_run_history CASCADE;
DROP TABLE IF EXISTS quality_alerts CASCADE;

CREATE TABLE customers (
    customer_id INTEGER,
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    email VARCHAR(255),
    phone VARCHAR(50),
    created_date DATE
);

CREATE TABLE orders (
    order_id INTEGER,
    customer_id INTEGER,
    order_date DATE,
    amount NUMERIC(10, 2),
    status VARCHAR(50)
);

CREATE TABLE data_quality_metrics (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100),
    check_type VARCHAR(100),
    total_records INTEGER,
    passed_records INTEGER,
    failed_records INTEGER,
    quality_score NUMERIC(5, 2),
    run_timestamp TIMESTAMP DEFAULT NOW()
);

CREATE TABLE quality_run_history (
    id SERIAL PRIMARY KEY,
    run_timestamp TIMESTAMP DEFAULT NOW(),
    overall_health_score NUMERIC(5, 2),
    tables_checked INTEGER,
    total_checks INTEGER,
    checks_passed INTEGER,
    checks_failed INTEGER
);

CREATE TABLE quality_alerts (
    id SERIAL PRIMARY KEY,
    alert_timestamp TIMESTAMP DEFAULT NOW(),
    table_name VARCHAR(100),
    check_type VARCHAR(100),
    threshold NUMERIC(5, 2),
    actual_score NUMERIC(5, 2),
    severity VARCHAR(20)
);
""")

What do these tables store?

  • customers and orders hold the operational data you will validate.
  • data_quality_metrics stores quality scores per table and check type after each pipeline run.
  • quality_run_history tracks overall health scores over time.
  • quality_alerts records threshold violations (used in the Secret Mission).
  • Below the table creation code, add the sample data inserts and connection close:
# Insert messy sample data into customers
cur.execute("""
INSERT INTO customers (customer_id, first_name, last_name, email, phone, created_date) VALUES
(1, 'Alice', 'Johnson', 'alice.johnson@email.com', '555-0101', '2024-01-15'),
(2, 'Bob', 'Smith', 'bob.smith@email.com', '555-0102', '2024-02-20'),
(3, NULL, 'Williams', 'invalid-email', '555-0103', '2024-03-10'),
(4, 'Diana', 'Brown', 'diana.brown@email.com', NULL, '2024-04-05'),
(5, 'Eve', 'Davis', 'eve.davis@email.com', '555-0105', '2024-05-12'),
(6, 'Frank', NULL, 'frank@email.com', '555-0106', '2024-06-18'),
(7, 'Grace', 'Miller', 'grace.miller@email.com', '555-0107', '2024-07-22'),
(8, 'Grace', 'Miller', 'grace.miller@email.com', '555-0107', '2024-07-22'),
(9, 'Henry', 'Wilson', NULL, '555-0109', '2024-08-30'),
(10, 'Ivy', 'Moore', 'ivy.moore@email.com', '555-0110', NULL),
(11, 'Jack', 'Taylor', 'jack.taylor@email.com', '555-0111', '2024-10-14'),
(12, NULL, NULL, NULL, NULL, NULL),
(13, 'Liam', 'Anderson', 'liam.anderson@email.com', '555-0113', '2024-12-01'),
(14, 'Mia', 'Thomas', 'MIA.THOMAS@EMAIL.COM', '555-0114', '2025-01-08'),
(15, 'Noah', 'Jackson', 'noah jackson@email.com', '555-0115', '2025-02-14');
""")

# Insert messy sample data into orders
cur.execute("""
INSERT INTO orders (order_id, customer_id, order_date, amount, status) VALUES
(1001, 1, '2024-03-01', 150.00, 'completed'),
(1002, 2, '2024-03-05', 89.99, 'completed'),
(1003, 3, '2024-03-10', -25.00, 'completed'),
(1004, 4, '2024-03-15', 200.50, 'pending'),
(1005, 5, '2024-03-20', 75.00, 'completed'),
(1006, NULL, '2024-03-25', 120.00, 'shipped'),
(1007, 7, NULL, 95.00, 'completed'),
(1008, 8, '2024-04-01', 999999.99, 'completed'),
(1009, 9, '2024-04-05', 45.00, 'pending'),
(1010, 10, '2024-04-10', NULL, 'completed'),
(1011, 11, '2024-04-15', 180.00, 'invalid_status'),
(1012, 99, '2024-04-20', 60.00, 'completed'),
(1013, 13, '2024-04-25', 110.00, 'completed'),
(1013, 13, '2024-04-25', 110.00, 'completed'),
(1015, 15, '2024-05-01', 250.00, 'shipped');
""")

cur.close()
conn.close()
print("Database setup complete. Tables created and sample data loaded.")

What quality issues are hiding in this data?

The customers table has NULL first names, an invalid email format (invalid-email), a space in an email (noah jackson@email.com), an exact duplicate row (customer 8), and a row that is entirely NULL.

The orders table has a negative amount, a wildly high amount (999999.99), NULL customer IDs, a duplicate order ID, and an invalid status value. These are the problems your validation pipeline will catch.

  • Save setup_database.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved both .env and setup_database.py.

ⓧ I'd like to double check the full code

Here is the complete setup_database.py file:

import psycopg2
from dotenv import load_dotenv
import os

load_dotenv()

conn = psycopg2.connect(
    host=os.getenv("DB_HOST"),
    port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD")
)
conn.autocommit = True
cur = conn.cursor()

# Create tables
cur.execute("""
DROP TABLE IF EXISTS orders CASCADE;
DROP TABLE IF EXISTS customers CASCADE;
DROP TABLE IF EXISTS data_quality_metrics CASCADE;
DROP TABLE IF EXISTS quality_run_history CASCADE;
DROP TABLE IF EXISTS quality_alerts CASCADE;

CREATE TABLE customers (
    customer_id INTEGER,
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    email VARCHAR(255),
    phone VARCHAR(50),
    created_date DATE
);

CREATE TABLE orders (
    order_id INTEGER,
    customer_id INTEGER,
    order_date DATE,
    amount NUMERIC(10, 2),
    status VARCHAR(50)
);

CREATE TABLE data_quality_metrics (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100),
    check_type VARCHAR(100),
    total_records INTEGER,
    passed_records INTEGER,
    failed_records INTEGER,
    quality_score NUMERIC(5, 2),
    run_timestamp TIMESTAMP DEFAULT NOW()
);

CREATE TABLE quality_run_history (
    id SERIAL PRIMARY KEY,
    run_timestamp TIMESTAMP DEFAULT NOW(),
    overall_health_score NUMERIC(5, 2),
    tables_checked INTEGER,
    total_checks INTEGER,
    checks_passed INTEGER,
    checks_failed INTEGER
);

CREATE TABLE quality_alerts (
    id SERIAL PRIMARY KEY,
    alert_timestamp TIMESTAMP DEFAULT NOW(),
    table_name VARCHAR(100),
    check_type VARCHAR(100),
    threshold NUMERIC(5, 2),
    actual_score NUMERIC(5, 2),
    severity VARCHAR(20)
);
""")

# Insert messy sample data into customers
cur.execute("""
INSERT INTO customers (customer_id, first_name, last_name, email, phone, created_date) VALUES
(1, 'Alice', 'Johnson', 'alice.johnson@email.com', '555-0101', '2024-01-15'),
(2, 'Bob', 'Smith', 'bob.smith@email.com', '555-0102', '2024-02-20'),
(3, NULL, 'Williams', 'invalid-email', '555-0103', '2024-03-10'),
(4, 'Diana', 'Brown', 'diana.brown@email.com', NULL, '2024-04-05'),
(5, 'Eve', 'Davis', 'eve.davis@email.com', '555-0105', '2024-05-12'),
(6, 'Frank', NULL, 'frank@email.com', '555-0106', '2024-06-18'),
(7, 'Grace', 'Miller', 'grace.miller@email.com', '555-0107', '2024-07-22'),
(8, 'Grace', 'Miller', 'grace.miller@email.com', '555-0107', '2024-07-22'),
(9, 'Henry', 'Wilson', NULL, '555-0109', '2024-08-30'),
(10, 'Ivy', 'Moore', 'ivy.moore@email.com', '555-0110', NULL),
(11, 'Jack', 'Taylor', 'jack.taylor@email.com', '555-0111', '2024-10-14'),
(12, NULL, NULL, NULL, NULL, NULL),
(13, 'Liam', 'Anderson', 'liam.anderson@email.com', '555-0113', '2024-12-01'),
(14, 'Mia', 'Thomas', 'MIA.THOMAS@EMAIL.COM', '555-0114', '2025-01-08'),
(15, 'Noah', 'Jackson', 'noah jackson@email.com', '555-0115', '2025-02-14');
""")

# Insert messy sample data into orders
cur.execute("""
INSERT INTO orders (order_id, customer_id, order_date, amount, status) VALUES
(1001, 1, '2024-03-01', 150.00, 'completed'),
(1002, 2, '2024-03-05', 89.99, 'completed'),
(1003, 3, '2024-03-10', -25.00, 'completed'),
(1004, 4, '2024-03-15', 200.50, 'pending'),
(1005, 5, '2024-03-20', 75.00, 'completed'),
(1006, NULL, '2024-03-25', 120.00, 'shipped'),
(1007, 7, NULL, 95.00, 'completed'),
(1008, 8, '2024-04-01', 999999.99, 'completed'),
(1009, 9, '2024-04-05', 45.00, 'pending'),
(1010, 10, '2024-04-10', NULL, 'completed'),
(1011, 11, '2024-04-15', 180.00, 'invalid_status'),
(1012, 99, '2024-04-20', 60.00, 'completed'),
(1013, 13, '2024-04-25', 110.00, 'completed'),
(1013, 13, '2024-04-25', 110.00, 'completed'),
(1015, 15, '2024-05-01', 250.00, 'shipped');
""")

cur.close()
conn.close()
print("Database setup complete. Tables created and sample data loaded.")
  • Run the setup script by executing this command in your terminal (make sure your virtual environment is active):
python setup_database.py

You should see: Database setup complete. Tables created and sample data loaded.

Connection refused or timeout?

This usually means the security group is not configured correctly. Go back to the RDS console and confirm the inbound rule allows port 5432 from your current IP address.

Also confirm the DB_HOST value in your .env file matches the endpoint from the RDS Connectivity & security tab exactly.

help me troubleshoot my RDS PostgreSQL connection

  • Verify the data loaded correctly by running this quick check in Python:
python -c "import psycopg2; from dotenv import load_dotenv; import os; load_dotenv(); conn = psycopg2.connect(host=os.getenv('DB_HOST'), port=os.getenv('DB_PORT'), dbname=os.getenv('DB_NAME'), user=os.getenv('DB_USER'), password=os.getenv('DB_PASSWORD')); cur = conn.cursor(); cur.execute('SELECT COUNT(*) FROM customers'); print(f'Customers: {cur.fetchone()[0]}'); cur.execute('SELECT COUNT(*) FROM orders'); print(f'Orders: {cur.fetchone()[0]}'); cur.close(); conn.close()"

You should see:

Customers: 15
Orders: 15

Both tables have 15 rows of messy data, ready for validation.

Your environment is ready and your database is loaded with messy operational data. Next up, you will build a validation pipeline using Great Expectations to automatically detect every quality issue hiding in those records.

Build Data Quality Validation with Great Expectations

Your database is loaded with messy data. Now you need a way to automatically catch what's wrong with it.

In this step, you'll use Great Expectations to define quality rules (called "expectations") and run them against your PostgreSQL tables. GX checks every row and tells you exactly which rules pass and which fail.

In this step, get ready to:
  • Connect Great Expectations to your RDS PostgreSQL instance and register tables as data assets.
  • Define expectation suites with quality rules for both the customers and orders tables.
  • Run validations and generate Data Docs showing pass/fail results.
Initialize Great Expectations and connect to PostgreSQL

Great Expectations needs a connection to your database before it can validate anything. You'll set up a Data Context (GX's configuration hub), point it at your RDS instance, and register both tables as assets.

  • In VS Code, create a new file called validate_data.py in your project folder.
  • Add the following code to set up the GX context and connect to your database:
import great_expectations as gx
from dotenv import load_dotenv
import os
import json

load_dotenv()

connection_string = (
    f"postgresql+psycopg2://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
    f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
)

# Initialize GX context
context = gx.get_context()

# Add PostgreSQL data source
datasource = context.data_sources.add_or_update_postgres(
    name="rds_postgres",
    connection_string=connection_string
)

# Create table assets
customers_asset = datasource.add_table_asset(
    name="customers_asset",
    table_name="customers"
)
orders_asset = datasource.add_table_asset(
    name="orders_asset",
    table_name="orders"
)

# Create batch definitions
customers_batch_def = customers_asset.add_batch_definition_whole_table(
    name="customers_full"
)
orders_batch_def = orders_asset.add_batch_definition_whole_table(
    name="orders_full"
)

What does this code do?

This block handles all the plumbing between Great Expectations and your PostgreSQL database.

  • connection_string builds the database URL from your .env variables using the format GX expects: postgresql+psycopg2://USER:PASSWORD@HOST:PORT/DATABASE.
  • gx.get_context() creates a Data Context, which is GX's central configuration object. It stores your data sources, suites, and validation results.
  • add_or_update_postgres() registers your RDS instance as a data source that GX can query directly.
  • add_table_asset() tells GX which specific tables to treat as validation targets.
  • add_batch_definition_whole_table() defines that each validation run should check the entire table, not a filtered subset.
  • Save validate_data.py.

You won't run this file yet. The next section adds the quality rules these assets will be validated against.

Define expectation suites

An expectation suite is a collection of rules that describe what "good" data looks like. You'll create one suite for customers and one for orders, each targeting the specific quality issues in your sample data.

What is an expectation suite?

Think of an expectation suite like a checklist for your data. Each expectation is a single rule ("this column must not be null", "values must be unique"). When you run the suite, GX checks every row against every rule and reports which passed and which failed.

  • In validate_data.py, add the following code below the batch definitions section:
# Define customers expectation suite
customers_suite = gx.ExpectationSuite(name="customers_quality_checks")
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="first_name")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="email")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="customer_id")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email",
        regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
    )
)
customers_suite = context.suites.add_or_update(customers_suite)

# Define orders expectation suite
orders_suite = gx.ExpectationSuite(name="orders_quality_checks")
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="amount")
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0, max_value=10000
    )
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
orders_suite = context.suites.add_or_update(orders_suite)

What do these expectations check?

The customers suite checks five things:

  • ExpectColumnValuesToNotBeNull on customer_id, first_name, and email ensures no critical fields are missing.
  • ExpectColumnValuesToBeUnique on customer_id catches duplicate records.
  • ExpectColumnValuesToMatchRegex on email verifies emails follow the standard format pattern.

The orders suite checks five things:

  • Not-null checks on order_id, customer_id, and amount catch missing required values.
  • ExpectColumnValuesToBeBetween ensures amounts fall between 0 and 10,000 (catching negatives and absurdly high values).
  • ExpectColumnValuesToBeUnique on order_id catches duplicate orders.
  • Save validate_data.py.

You won't see output yet. The next section creates validation definitions that pair these suites with your table batches and runs them.

Run validations and review results

Validation definitions link your expectation suites to specific batch definitions. Once linked, you can run them and GX will execute each expectation against your live data.

  • In validate_data.py, add the following code below the orders suite section:
# Create validation definitions
customers_validation = context.validation_definitions.add_or_update(
    gx.ValidationDefinition(
        name="customers_validation",
        data=customers_batch_def,
        suite=customers_suite
    )
)
orders_validation = context.validation_definitions.add_or_update(
    gx.ValidationDefinition(
        name="orders_validation",
        data=orders_batch_def,
        suite=orders_suite
    )
)

# Run validations
print("Running customers validation...")
customers_results = customers_validation.run()
print(f"  Success: {customers_results.success}")

print("Running orders validation...")
orders_results = orders_validation.run()
print(f"  Success: {orders_results.success}")

# Build Data Docs
context.build_data_docs()
print("\nData Docs built. Check gx/uncommitted/data_docs/local_site/index.html")

# Save results for the cleansing pipeline
results_summary = {
    "customers": {
        "success": customers_results.success,
        "statistics": customers_results.statistics
    },
    "orders": {
        "success": orders_results.success,
        "statistics": orders_results.statistics
    }
}

with open("validation_results.json", "w") as f:
    json.dump(results_summary, f, indent=2, default=str)

print("\nValidation results saved to validation_results.json")

What does this code do?

  • gx.ValidationDefinition pairs a batch definition (which table data to check) with a suite (which rules to apply). This pairing is what makes a validation runnable.
  • .run() executes all expectations in the suite against the live data in PostgreSQL and returns a results object.
  • context.build_data_docs() generates an HTML report you can open in your browser to see a visual breakdown of every expectation's pass/fail status.
  • The results are also saved to validation_results.json so the cleansing pipeline in the next step can read which checks failed.
  • Save validate_data.py.
  • Run the validation script by running this command in your terminal:
python validate_data.py
  • You should see output showing both validations completed with Success: False for each table. This is expected because your sample data intentionally contains quality issues.
  • Open the Data Docs report by navigating to gx/uncommitted/data_docs/local_site/index.html in your browser. You should see a detailed breakdown of which expectations passed and which failed.

Seeing a connection error?

  • Verify your .env file has the correct RDS endpoint, port, database name, username, and password.
  • Check that your RDS security group still allows inbound traffic on port 5432 from your current IP address. Your IP may have changed since Step 1.
  • Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.

Help me debug a Great Expectations PostgreSQL connection error

✔️ Awesome, I've got everything!

Great. Make sure you've saved validate_data.py and confirmed both validations returned Success: False.

ⓧ I'd like to double check the full code

Here is the complete validate_data.py file:

import great_expectations as gx
from dotenv import load_dotenv
import os
import json

load_dotenv()

connection_string = (
    f"postgresql+psycopg2://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
    f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
)

# Initialize GX context
context = gx.get_context()

# Add PostgreSQL data source
datasource = context.data_sources.add_or_update_postgres(
    name="rds_postgres",
    connection_string=connection_string
)

# Create table assets
customers_asset = datasource.add_table_asset(
    name="customers_asset",
    table_name="customers"
)
orders_asset = datasource.add_table_asset(
    name="orders_asset",
    table_name="orders"
)

# Create batch definitions
customers_batch_def = customers_asset.add_batch_definition_whole_table(
    name="customers_full"
)
orders_batch_def = orders_asset.add_batch_definition_whole_table(
    name="orders_full"
)

# Define customers expectation suite
customers_suite = gx.ExpectationSuite(name="customers_quality_checks")
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="first_name")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="email")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="customer_id")
)
customers_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email",
        regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
    )
)
customers_suite = context.suites.add_or_update(customers_suite)

# Define orders expectation suite
orders_suite = gx.ExpectationSuite(name="orders_quality_checks")
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="customer_id")
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="amount")
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0, max_value=10000
    )
)
orders_suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
orders_suite = context.suites.add_or_update(orders_suite)

# Create validation definitions
customers_validation = context.validation_definitions.add_or_update(
    gx.ValidationDefinition(
        name="customers_validation",
        data=customers_batch_def,
        suite=customers_suite
    )
)
orders_validation = context.validation_definitions.add_or_update(
    gx.ValidationDefinition(
        name="orders_validation",
        data=orders_batch_def,
        suite=orders_suite
    )
)

# Run validations
print("Running customers validation...")
customers_results = customers_validation.run()
print(f"  Success: {customers_results.success}")

print("Running orders validation...")
orders_results = orders_validation.run()
print(f"  Success: {orders_results.success}")

# Build Data Docs
context.build_data_docs()
print("\nData Docs built. Check gx/uncommitted/data_docs/local_site/index.html")

# Save results for the cleansing pipeline
results_summary = {
    "customers": {
        "success": customers_results.success,
        "statistics": customers_results.statistics
    },
    "orders": {
        "success": orders_results.success,
        "statistics": orders_results.statistics
    }
}

with open("validation_results.json", "w") as f:
    json.dump(results_summary, f, indent=2, default=str)

print("\nValidation results saved to validation_results.json")

Your validation pipeline is working. It found real quality issues in the data. Next up, you'll build a cleansing script that fixes these problems and tracks quality scores over time.

Automate Data Cleansing and Quality Scoring

Your validation pipeline has identified problems in the data. Now you need to act on those findings.

In this step, you'll build an automated cleansing script that calculates quality scores for every check type, logs those scores to the database, and then fixes the problems it finds. This creates a feedback loop where each pipeline run produces measurable improvements.

In this step, get ready to:
  • Build a reusable quality scoring function that checks records against rules and logs results.
  • Calculate and store quality metrics for both tables with timestamps for trend tracking.
  • Run the full pipeline end-to-end and observe improved scores after cleansing.
Build a cleansing pipeline

The cleansing pipeline needs a helper function that runs a SQL check, counts failures, and logs the result. This single function powers all 8 quality checks across both tables.

  • Create a new file called cleanse_and_report.py in your project folder.
  • Add the connection setup and the calculate_quality_score helper function by pasting this code:
import psycopg2
from dotenv import load_dotenv
import os
from datetime import datetime

load_dotenv()

conn = psycopg2.connect(
    host=os.getenv("DB_HOST"),
    port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD")
)
conn.autocommit = True
cur = conn.cursor()


def calculate_quality_score(table_name, check_type, check_query):
    """Run a quality check and return the score."""
    cur.execute(f"SELECT COUNT(*) FROM {table_name}")
    total = cur.fetchone()[0]

    cur.execute(check_query)
    failed = cur.fetchone()[0]

    passed = total - failed
    score = (passed / total * 100) if total > 0 else 0

    cur.execute("""
        INSERT INTO data_quality_metrics
            (table_name, check_type, total_records, passed_records, failed_records, quality_score)
        VALUES (%s, %s, %s, %s, %s, %s)
    """, (table_name, check_type, total, passed, failed, round(score, 2)))

    return score

What does this code do?

The calculate_quality_score function is the engine of the pipeline. It takes a table name, a check type label, and a SQL query that counts failures.

  • It first counts the total records in the table.
  • It then runs the failure query to find how many records break the rule.
  • It calculates a percentage score and inserts a row into data_quality_metrics with all the details.

Now add the quality checks for the customers table. Each call uses a SQL query that identifies records violating a specific rule.

  • Add the following code below the calculate_quality_score function:
# Run quality checks for customers table
print("Calculating quality scores for customers...")
scores = []

score = calculate_quality_score(
    "customers", "null_first_name",
    "SELECT COUNT(*) FROM customers WHERE first_name IS NULL"
)
scores.append(score)
print(f"  Null first_name check: {score:.1f}%")

score = calculate_quality_score(
    "customers", "null_email",
    "SELECT COUNT(*) FROM customers WHERE email IS NULL"
)
scores.append(score)
print(f"  Null email check: {score:.1f}%")

score = calculate_quality_score(
    "customers", "duplicate_customer_id",
    "SELECT COUNT(*) - COUNT(DISTINCT customer_id) FROM customers"
)
scores.append(score)
print(f"  Duplicate customer_id check: {score:.1f}%")

score = calculate_quality_score(
    "customers", "invalid_email_format",
    "SELECT COUNT(*) FROM customers WHERE email IS NOT NULL AND email !~ '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'"
)
scores.append(score)
print(f"  Invalid email format check: {score:.1f}%")

What do these checks catch?

  • null_first_name catches customers missing a first name (rows 3 and 12 in the sample data).
  • null_email catches customers with no email at all (rows 9 and 12).
  • duplicate_customer_id catches rows that share an ID (the duplicated Grace Miller record).
  • invalid_email_format uses a regex to find malformed emails like invalid-email and noah jackson@email.com (space in address).

Next, add the quality checks for the orders table below the customers checks.

  • Add this code directly below the last customers check:
# Run quality checks for orders table
print("\nCalculating quality scores for orders...")

score = calculate_quality_score(
    "orders", "null_customer_id",
    "SELECT COUNT(*) FROM orders WHERE customer_id IS NULL"
)
scores.append(score)
print(f"  Null customer_id check: {score:.1f}%")

score = calculate_quality_score(
    "orders", "null_amount",
    "SELECT COUNT(*) FROM orders WHERE amount IS NULL"
)
scores.append(score)
print(f"  Null amount check: {score:.1f}%")

score = calculate_quality_score(
    "orders", "amount_out_of_range",
    "SELECT COUNT(*) FROM orders WHERE amount IS NOT NULL AND (amount < 0 OR amount > 10000)"
)
scores.append(score)
print(f"  Amount out of range check: {score:.1f}%")

score = calculate_quality_score(
    "orders", "duplicate_order_id",
    "SELECT COUNT(*) - COUNT(DISTINCT order_id) FROM orders"
)
scores.append(score)
print(f"  Duplicate order_id check: {score:.1f}%")

What do the orders checks catch?

  • null_customer_id finds orphaned orders with no customer link (row 1006).
  • null_amount finds orders with missing monetary values (row 1010).
  • amount_out_of_range catches negative amounts and implausibly large ones like 999999.99.
  • duplicate_order_id finds repeated order IDs (order 1013 appears twice).
  • Save cleanse_and_report.py.

Seeing a syntax error in the regex?

The email regex uses double backslashes (\\.) because the backslash needs escaping inside the SQL string. Make sure you have exactly two backslashes before the period.

Help me fix a regex escaping issue in my cleansing pipeline.

Create quality metrics tables and log scores

The data_quality_metrics table was created in your setup script. Now you need to calculate an overall health score from all 8 checks and log it to the quality_run_history table for trend tracking.

  • Add the overall health calculation and history logging below the orders checks:
# Calculate overall health score
overall_health = sum(scores) / len(scores) if scores else 0

# Log to run history
cur.execute("""
    INSERT INTO quality_run_history
        (overall_health_score, tables_checked, total_checks, checks_passed, checks_failed)
    VALUES (%s, %s, %s, %s, %s)
""", (
    round(overall_health, 2),
    2,
    len(scores),
    sum(1 for s in scores if s >= 90),
    sum(1 for s in scores if s < 90)
))

print(f"\nOverall data health score: {overall_health:.1f}%")
print("Quality metrics logged to database.")

How is the health score calculated?

The overall health score is a simple average of all 8 individual check scores. A check is counted as "passed" if its score is 90% or above. This gives you a single number representing your data's overall quality.

The quality_run_history table stores one row per pipeline run. Over time, this lets you see whether your data quality is improving or degrading.

Now add the cleansing operations that fix the problems the checks identified. These run after scoring so you can compare before and after.

  • Add the cleansing logic and connection close below the health score output:
# Data cleansing operations
print("\nRunning data cleansing...")

# Standardize email to lowercase
cur.execute("UPDATE customers SET email = LOWER(email) WHERE email IS NOT NULL")
print("  Standardized emails to lowercase")

# Remove exact duplicate rows from customers
cur.execute("""
    DELETE FROM customers
    WHERE ctid NOT IN (
        SELECT MIN(ctid)
        FROM customers
        GROUP BY customer_id, first_name, last_name, email, phone, created_date
    )
""")
print(f"  Removed duplicate customer records")

# Remove exact duplicate rows from orders
cur.execute("""
    DELETE FROM orders
    WHERE ctid NOT IN (
        SELECT MIN(ctid)
        FROM orders
        GROUP BY order_id, customer_id, order_date, amount, status
    )
""")
print(f"  Removed duplicate order records")

print("\nCleansing complete. Run validate_data.py again to see improved scores.")

cur.close()
conn.close()

What does each cleansing step fix?

  • LOWER(email) standardizes mixed-case emails like MIA.THOMAS@EMAIL.COM to lowercase.
  • The ctid trick is PostgreSQL's way of identifying physical row locations. By keeping only the row with the smallest ctid for each group of identical values, exact duplicates are removed.
  • Records with null required fields are left in place but flagged by the scoring. The focus here is on fixable issues like case and duplicates.
  • Save cleanse_and_report.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before running it.

ⓧ I'd like to double check the full code

Here is the complete cleanse_and_report.py file:

import psycopg2
from dotenv import load_dotenv
import os
from datetime import datetime

load_dotenv()

conn = psycopg2.connect(
    host=os.getenv("DB_HOST"),
    port=os.getenv("DB_PORT"),
    dbname=os.getenv("DB_NAME"),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD")
)
conn.autocommit = True
cur = conn.cursor()


def calculate_quality_score(table_name, check_type, check_query):
    """Run a quality check and return the score."""
    cur.execute(f"SELECT COUNT(*) FROM {table_name}")
    total = cur.fetchone()[0]

    cur.execute(check_query)
    failed = cur.fetchone()[0]

    passed = total - failed
    score = (passed / total * 100) if total > 0 else 0

    cur.execute("""
        INSERT INTO data_quality_metrics
            (table_name, check_type, total_records, passed_records, failed_records, quality_score)
        VALUES (%s, %s, %s, %s, %s, %s)
    """, (table_name, check_type, total, passed, failed, round(score, 2)))

    return score


# Run quality checks for customers table
print("Calculating quality scores for customers...")
scores = []

score = calculate_quality_score(
    "customers", "null_first_name",
    "SELECT COUNT(*) FROM customers WHERE first_name IS NULL"
)
scores.append(score)
print(f"  Null first_name check: {score:.1f}%")

score = calculate_quality_score(
    "customers", "null_email",
    "SELECT COUNT(*) FROM customers WHERE email IS NULL"
)
scores.append(score)
print(f"  Null email check: {score:.1f}%")

score = calculate_quality_score(
    "customers", "duplicate_customer_id",
    "SELECT COUNT(*) - COUNT(DISTINCT customer_id) FROM customers"
)
scores.append(score)
print(f"  Duplicate customer_id check: {score:.1f}%")

score = calculate_quality_score(
    "customers", "invalid_email_format",
    "SELECT COUNT(*) FROM customers WHERE email IS NOT NULL AND email !~ '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'"
)
scores.append(score)
print(f"  Invalid email format check: {score:.1f}%")

# Run quality checks for orders table
print("\nCalculating quality scores for orders...")

score = calculate_quality_score(
    "orders", "null_customer_id",
    "SELECT COUNT(*) FROM orders WHERE customer_id IS NULL"
)
scores.append(score)
print(f"  Null customer_id check: {score:.1f}%")

score = calculate_quality_score(
    "orders", "null_amount",
    "SELECT COUNT(*) FROM orders WHERE amount IS NULL"
)
scores.append(score)
print(f"  Null amount check: {score:.1f}%")

score = calculate_quality_score(
    "orders", "amount_out_of_range",
    "SELECT COUNT(*) FROM orders WHERE amount IS NOT NULL AND (amount < 0 OR amount > 10000)"
)
scores.append(score)
print(f"  Amount out of range check: {score:.1f}%")

score = calculate_quality_score(
    "orders", "duplicate_order_id",
    "SELECT COUNT(*) - COUNT(DISTINCT order_id) FROM orders"
)
scores.append(score)
print(f"  Duplicate order_id check: {score:.1f}%")

# Calculate overall health score
overall_health = sum(scores) / len(scores) if scores else 0

# Log to run history
cur.execute("""
    INSERT INTO quality_run_history
        (overall_health_score, tables_checked, total_checks, checks_passed, checks_failed)
    VALUES (%s, %s, %s, %s, %s)
""", (
    round(overall_health, 2),
    2,
    len(scores),
    sum(1 for s in scores if s >= 90),
    sum(1 for s in scores if s < 90)
))

print(f"\nOverall data health score: {overall_health:.1f}%")
print("Quality metrics logged to database.")

# Data cleansing operations
print("\nRunning data cleansing...")

# Standardize email to lowercase
cur.execute("UPDATE customers SET email = LOWER(email) WHERE email IS NOT NULL")
print("  Standardized emails to lowercase")

# Remove exact duplicate rows from customers
cur.execute("""
    DELETE FROM customers
    WHERE ctid NOT IN (
        SELECT MIN(ctid)
        FROM customers
        GROUP BY customer_id, first_name, last_name, email, phone, created_date
    )
""")
print(f"  Removed duplicate customer records")

# Remove exact duplicate rows from orders
cur.execute("""
    DELETE FROM orders
    WHERE ctid NOT IN (
        SELECT MIN(ctid)
        FROM orders
        GROUP BY order_id, customer_id, order_date, amount, status
    )
""")
print(f"  Removed duplicate order records")

print("\nCleansing complete. Run validate_data.py again to see improved scores.")

cur.close()
conn.close()
Run the full pipeline end-to-end

Time to execute the complete workflow and verify that metrics are being written to the database.

  • Run the cleansing pipeline by executing this command in your terminal:
python cleanse_and_report.py
  • You should see output listing each check with its quality score percentage, an overall health score, and confirmation that cleansing operations completed.

Connection refused or authentication error?

Make sure your RDS instance is still running and your .env file has the correct endpoint, username, and password. Check that your security group still allows inbound traffic on port 5432 from your current IP address.

Help me troubleshoot my PostgreSQL connection error in cleanse_and_report.py.

Now verify that the metrics were stored in the database. Run this quick check:

  • Confirm metrics exist by running this command:
python -c "import psycopg2; from dotenv import load_dotenv; import os; load_dotenv(); conn = psycopg2.connect(host=os.getenv('DB_HOST'), port=os.getenv('DB_PORT'), dbname=os.getenv('DB_NAME'), user=os.getenv('DB_USER'), password=os.getenv('DB_PASSWORD')); cur = conn.cursor(); cur.execute('SELECT COUNT(*) FROM data_quality_metrics'); print(f'Metrics rows: {cur.fetchone()[0]}'); cur.execute('SELECT COUNT(*) FROM quality_run_history'); print(f'History rows: {cur.fetchone()[0]}'); cur.close(); conn.close()"
  • You should see Metrics rows: 8 and History rows: 1.

Why 8 metrics rows?

Each call to calculate_quality_score inserts one row into data_quality_metrics. You ran 4 checks for customers and 4 checks for orders, producing 8 rows total.

Run the pipeline a second time to demonstrate trend tracking. Since the first run already cleansed duplicates and standardized emails, the second run should produce higher scores.

  • Run the pipeline again by executing:
python cleanse_and_report.py
  • Compare the output to the first run. The duplicate checks should now show 100% because the first run already removed duplicates.
  • Verify the trend data exists by running the row count check again:
python -c "import psycopg2; from dotenv import load_dotenv; import os; load_dotenv(); conn = psycopg2.connect(host=os.getenv('DB_HOST'), port=os.getenv('DB_PORT'), dbname=os.getenv('DB_NAME'), user=os.getenv('DB_USER'), password=os.getenv('DB_PASSWORD')); cur = conn.cursor(); cur.execute('SELECT COUNT(*) FROM data_quality_metrics'); print(f'Metrics rows: {cur.fetchone()[0]}'); cur.execute('SELECT COUNT(*) FROM quality_run_history'); print(f'History rows: {cur.fetchone()[0]}'); cur.close(); conn.close()"
  • You should now see Metrics rows: 16 (8 from each run) and History rows: 2.

Why run it twice?

Running the pipeline twice demonstrates trend tracking in action. The quality_run_history table now has two timestamped snapshots of your overall data health. This is exactly what your QuickSight dashboard will visualize in the next step.

Your automated cleansing pipeline is working. It scores, logs, and fixes data quality issues in a single run. Next up, you'll connect this data to Amazon QuickSight to build a live dashboard that visualizes these quality trends.

Build a QuickSight Dashboard

Your data quality pipeline is running, quality scores are logged, and your PostgreSQL database holds a history of data health metrics. The missing piece is visibility.

Stakeholders and team leads need to see data health at a glance without running SQL queries. In this step, you will connect Amazon QuickSight to your RDS database and build an interactive dashboard displaying quality KPIs.

In this step, get ready to:
  • Sign up for Amazon QuickSight and connect it to your RDS PostgreSQL instance.
  • Build four data quality visualizations in a dashboard.
  • Publish the dashboard and verify it reflects live pipeline data.
Set up Amazon QuickSight and connect to PostgreSQL

Amazon QuickSight is a cloud-native BI service that connects directly to AWS data sources. You will sign up for the 30-day free trial and point it at your RDS instance.

  • Open the AWS Management Console in your browser.
  • In the search bar at the top, type QuickSight and select the QuickSight service.
  • Click Sign up for QuickSight.
  • Select the Enterprise edition (required for VPC connectivity to RDS).
  • Enter your email address for notifications.
  • For Authentication method, select Use IAM federated identities & QuickSight-managed users.
  • Enter a unique QuickSight account name (e.g. dataqualityproject).
  • Click Finish.

What is the QuickSight free trial?

Amazon QuickSight offers a 30-day free trial with all features for up to 25 users per account. After the trial ends, the Author tier costs $24 per user per month.

You will deactivate QuickSight in the Before You Go section to avoid any charges.

Once QuickSight finishes setting up, you will land on the QuickSight home page.

Before QuickSight can reach your RDS database, you need to allow its IP range through your RDS security group.

  • Open a new browser tab and navigate to the VPC console (search for VPC in the AWS Console search bar).
  • Click Security Groups in the left navigation pane.
  • Click Create security group.
  • For Security group name and Description, enter Amazon-QuickSight-access.
  • For VPC, select the same VPC that your RDS instance uses.
  • Under Inbound rules, click Add rule.
  • Set Type to Custom TCP, Port range to 5432, and Source to the QuickSight CIDR block for your AWS region.
  • Click Create security group.

How do I find my region's QuickSight CIDR?

Each AWS region has a specific IP range that QuickSight uses to connect to data sources. Common examples:

  • US East (N. Virginia): 52.23.63.224/27
  • US West (Oregon): 54.70.204.128/27
  • Europe (Ireland): 52.210.255.224/27
  • Europe (Frankfurt): 35.158.127.192/27

Find your region's full list at the AWS Regions, websites, IP address ranges, and endpoints page in the QuickSight docs.

  • Now attach this security group to your RDS instance. Navigate to the RDS console.
  • Click Databases in the left navigation, then select your database instance.
  • Click Modify.
  • Scroll to the Connectivity section. Under Security group, add Amazon-QuickSight-access alongside your existing security group.
  • Click Continue, then click Modify DB instance (choose Apply immediately).

Your RDS instance now accepts connections from QuickSight.

  • Switch back to the QuickSight tab in your browser.
  • Click Datasets in the left navigation pane.
  • Click New dataset.
  • Select PostgreSQL as the data source type.
  • Enter a data source name (e.g. DataQualityRDS).
  • For Database server, paste your RDS endpoint (the same value as DB_HOST in your .env file).
  • For Port, enter 5432.
  • For Database name, enter dataqualitydb.
  • For Username, enter postgres.
  • For Password, enter the master password you set when creating the RDS instance.
  • Click Validate connection. Wait for the green success message.
  • Click Create data source.

Connection validation failed?

  • Double-check that the Amazon-QuickSight-access security group is attached to your RDS instance (not just created).
  • Confirm the CIDR you entered matches the region where you signed up for QuickSight.
  • Verify your RDS instance has Public accessibility set to Yes (required for non-VPC QuickSight connections).

Still stuck? Help me troubleshoot my QuickSight connection to RDS PostgreSQL

Create data quality KPI visualizations

With your data source connected, you will create a dataset from your metrics tables and build four visualizations that show data health at a glance.

  • After creating the data source, QuickSight prompts you to choose a table. Select data_quality_metrics and click Select.
  • Choose Directly query your data (this keeps visualizations live from your database).
  • Click Visualize to open the analysis editor.

You are now in the QuickSight analysis view. You will build four charts in this single analysis.

First, create an overall data health KPI:

  • In the Visual types panel (bottom left), select the KPI chart type.
  • Drag quality_score to the Value field well.
  • Click the dropdown arrow on the field and change the aggregation to Average.
  • Give the visual a title: click the title text and type Overall Data Health Score.

Next, add a bar chart showing pass/fail rates by check type:

  • Click Add at the top of the analysis, then select Add visual.
  • Select the Vertical bar chart visual type.
  • Drag check_type to the X axis field well.
  • Drag passed_records to the Value field well.
  • Drag failed_records to the Value field well (this creates a grouped bar).
  • Title the visual Pass/Fail Rates by Check Type.

Now create a line chart showing quality trends over time:

  • Add another visual. Select the Line chart type.
  • Drag run_timestamp to the X axis.
  • Drag quality_score to the Value field well. Set the aggregation to Average.
  • Title the visual Quality Score Trends Over Time.

Finally, create a grouped bar chart comparing quality across tables:

  • Add one more visual. Select Vertical bar chart.
  • Drag table_name to the X axis.
  • Drag quality_score to the Value field well (aggregated as Average).
  • Drag check_type to the Group/Color field well to create a grouped bar chart.
  • Title the visual Table-Level Quality Comparison.

Arrange the four visuals by dragging them into a 2x2 grid layout in the analysis canvas.

Why directly query instead of SPICE?

QuickSight offers two modes: SPICE (in-memory cache) and direct query. Direct query always shows the latest data from your database without needing a manual refresh.

For a data quality dashboard that needs to reflect the most recent pipeline run, direct query is the better fit.

Publish and share the dashboard

An analysis is your working space. A dashboard is the published, shareable version that other users can view.

  • Before publishing, add interactive filters. Click Filter in the left toolbar.
  • Click Add filter, select run_timestamp, and choose a Date range filter type.
  • Add a second filter on table_name as a Filter list type.
  • For each filter, click the three-dot menu and select Add to sheet to make it a visible control on the dashboard.

Now publish the analysis as a dashboard:

  • Click the Share icon in the top-right corner of the analysis.
  • Select Publish dashboard.
  • Enter a dashboard name (e.g. Data Quality KPIs).
  • Click Publish dashboard.

QuickSight opens the published dashboard. You should see your four visualizations with the date range and table name filter controls at the top.

To verify the dashboard updates with fresh data, run the pipeline again:

  • Switch back to your terminal in VS Code.
  • Run the cleansing pipeline again by running:
python cleanse_and_report.py
  • Return to the QuickSight dashboard in your browser.
  • Click the refresh icon or reload the page. The KPI and trend chart should now reflect the additional data point from your latest pipeline run.

Dashboard not showing updated data?

  • Make sure you chose Directly query your data when creating the dataset. If you chose SPICE, you need to manually refresh the dataset.
  • Confirm cleanse_and_report.py ran without errors and printed the updated health score.
  • Check that your date range filter is not excluding the latest run timestamp.

Need more help? Why isn't my QuickSight dashboard showing the latest data from my RDS PostgreSQL database?

Your data quality system now has a professional reporting layer. Next up, you will add threshold-based alerting in the Secret Mission to catch quality drops before they become problems.

Secret mission

Add Threshold Alerting and Summary Reports

Your pipeline validates and cleanses data, but nobody gets notified when quality drops. In this mission, you'll add threshold-based alerting that flags when scores fall below acceptable levels. You'll also generate a formatted summary report that a data team lead could read at a glance.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses an Amazon RDS instance (Free Tier eligible) and an Amazon QuickSight account (30-day free trial), both of which can incur charges if left active.

Cost warning

Your RDS db.t3.micro instance is covered by the AWS Free Tier for 750 hours per month (legacy accounts) or included in Free Plan credits (accounts created after July 15, 2025). After your free tier expires, you will be charged for the running instance. Your QuickSight account starts a 30-day free trial. After 30 days, it costs $24 per author per month. Delete or unsubscribe before the trial ends to avoid charges.

Resources you used:

  • Amazon RDS PostgreSQL instance (db.t3.micro, 20 GB gp2 storage).
  • Amazon QuickSight account (30-day free trial).
  • Local project folder with Python scripts, virtual environment, and GX data context.

Keep everything running

No action needed. Choose this if you are still actively building or want to keep experimenting with your data quality pipeline.

  • Your RDS instance continues running. Monitor your Free Tier usage in the AWS Billing console to avoid surprise charges.
  • Your QuickSight account remains active. Set a calendar reminder to cancel before your 30-day trial expires if you do not plan to keep using it.
  • Your local project files and virtual environment remain on your machine.

Pause - I'll come back to this later

Stop your RDS instance to eliminate compute charges while keeping your data intact. Storage charges still apply but are covered by the Free Tier.

  • Open the Amazon RDS console at https://console.aws.amazon.com/rds/.
  • In the navigation pane, choose Databases, then select your DB instance.
  • Choose Actions, then Stop temporarily.
  • Acknowledge that the instance will automatically restart after 7 days, then click Stop temporarily.
  • Set a calendar reminder for your QuickSight 30-day trial deadline. Cancel before it expires to avoid charges.
  • Deactivate your Python virtual environment by running:
deactivate

Why does the instance restart after 7 days?

AWS automatically restarts stopped RDS instances after 7 consecutive days to apply pending maintenance updates. If you need to keep it stopped longer, stop it again after it restarts.

Delete - I don't want to use this again

Remove all project resources completely to ensure $0 charges going forward.

Delete your RDS PostgreSQL instance
  • Open the Amazon RDS console at https://console.aws.amazon.com/rds/.
  • In the navigation pane, choose Databases, then select your DB instance.
  • Choose Actions, then Delete.
  • Uncheck Create final snapshot? (you do not need a snapshot for this tutorial data).
  • Enter delete me in the confirmation box.
  • Click Delete.
Unsubscribe from Amazon QuickSight
  • Sign in to AWS and navigate to the QuickSight unsubscribe page: https://us-east-1.quicksight.aws.amazon.com/sn/console/unsubscribe
  • Follow the on-screen prompts to confirm account deletion. This deletes all QuickSight data, analyses, and users.

QuickSight deletion is permanent

Terminating your QuickSight account is instant and cannot be undone. All dashboards and datasets will be permanently removed.

Delete local project files
  • Deactivate your virtual environment first by running:
deactivate
  • Delete the project folder from your Desktop by running:

macOS

cd ~/Desktop
rm -rf data-quality-pipeline

Windows

cd ~\Desktop
rmdir /s /q data-quality-pipeline

Nice Work!

Nice Work!

You just built a complete data quality management system from scratch. Your pipeline validates messy data, cleanses it automatically, and surfaces health metrics through a live dashboard.

You've learned how to:

  • Build an automated data quality validation pipeline using Great Expectations that checks for nulls, duplicates, format violations, and range errors in a PostgreSQL database.
  • Automate data cleansing that standardizes formats, removes duplicates, and logs quality scores to metrics tables for trend tracking over time.
  • Build an interactive Amazon QuickSight dashboard displaying data quality KPIs connected to your Amazon RDS PostgreSQL instance.
  • Secret Mission: Added threshold-based alerting that generates summary reports when quality scores drop below acceptable levels.

Ready to quiz yourself?