Build an AI Data Quality Pipeline

Turn messy CSV orders into validated data and load them into SQLite.

Introduction

30 Second Summary

A spreadsheet can look tidy while hiding duplicate orders. Blank fields or broken prices can quietly distort every report built from it.

In this project, you will build a browser-based retail-orders data-quality pipeline in Google Colab. Deterministic Python rules control which records pass while Colab AI explains the results.

What You'll Build

Picture running one notebook to watch nine messy orders become four trusted records plus five explained rejections before the clean data lands in a queryable SQLite warehouse.

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

  • A repeatable ingestion demo that turns a synthetic nine-row CSV file into dictionary-shaped records you can inspect.
  • A visible quarantine workflow that keeps four valid orders while showing clear rejection reasons for five invalid orders.
  • An auditable quality report saved as JSON with an AI-generated briefing based only on the calculated metrics.
  • Secret Mission: Add a cross-layer reconciliation gate that compares order identities plus revenue between the clean pipeline output and SQLite.

Are there any prerequisites?

You'll need a Google Account with Colab AI access. Eligibility requires an account age of 18+ plus a supported locale.

You do not need prior Python or SQL experience. Google Colab runs entirely in your browser.

Before We Start

This is your moment to lock in what you are building before the hands-on work begins. AI helps explain the results. Deterministic rules control which retail orders reach the warehouse.

Set Up Colab and Verify AI

Your retail-orders pipeline needs a workspace where every check can be run visibly.

Google Colab provides that hosted workspace in your browser. This step also confirms Colab AI works before later notebook cells depend on it.

In this step, get ready to:
  • Create a named Google Colab notebook in the default hosted runtime.
  • Confirm the Python runtime through a visible cell output.
  • Verify both Colab AI entry points with a synthetic ETL prompt.
Create the Colab notebook

Google Colab runs notebook code in a hosted browser workspace. This pipeline only needs the default runtime.

Your existing Google account is the only sign-in this step needs. No billing account or API key is involved.

  • Open Google Colab in a new browser tab.
  • Sign in with the Google account you confirmed for this project.
  • Create a blank notebook from the Colab welcome screen.
  • Select the notebook title in the top-left corner.
  • Replace the current title with AI Data Quality Pipeline.
  • Leave the notebook on its default hosted runtime.

You'll see an empty notebook titled AI Data Quality Pipeline with space for your first code cell.

Why use a hosted notebook?

Google Colab provides the notebook editor plus a hosted runtime. You can focus on the pipeline without installing Python or an IDE on your Windows device.

Check the Python workspace

A code cell holds instructions for the hosted Python runtime. Its output area shows the result directly underneath.

  • Create a code cell from the notebook toolbar.
  • Enter the readiness check by copying the code below:
print("Pipeline workspace ready")

What does this check prove?

  • The print() function sends a fixed message to the cell output.
  • Seeing that message confirms the hosted Python runtime executed your code.

Before you run the cell, where do you expect its message to appear?

  • Select the triangular run button on the left side of the code cell.

You'll see Pipeline workspace ready in the output area beneath the cell.

The editable box is the code cell. The triangular control beside it is the run button.

The result underneath is the output area.

No readiness message?

  • Confirm the cell contains the complete print("Pipeline workspace ready") line.
  • Run the cell again after the hosted runtime finishes connecting.

Help me understand why my Colab code cell is not showing the readiness message.

Verify both AI entry points

The next check uses ETL as a harmless synthetic topic. The footer chat gives you a conversational AI entry point.

AI supplies explanations in this project. Deterministic code retains authority over which records reach the warehouse.

Keep AI prompts synthetic

Colab AI prompts can be processed as part of the service. This project uses invented retail data so no personal or confidential information needs to enter a prompt.

  • Keep every Colab AI prompt limited to synthetic project information.
  • Select the Gemini spark icon in the notebook footer.
  • Enter explain ETL in beginner language in the chat panel.
  • Use the chat panel's send control to submit the prompt.

You'll see a beginner-friendly explanation of ETL. Its exact wording can vary.

  • Create another code cell beneath the readiness check.
  • Verify the Python AI interface by copying the code below:
# Import Colab's hosted AI interface
from google.colab import ai

# Print a short response to a synthetic ETL prompt
print(ai.generate_text("In one short sentence, explain what ETL means."))

What does this code do?

  • The from google.colab import ai line makes Colab's hosted AI interface available in the cell.
  • The ai.generate_text() call sends the synthetic ETL prompt.
  • The outer print() call places the returned text in the notebook output.

Before you run this cell, do you expect a fixed message or a generated explanation?

  • Select the run button beside the AI verification cell.

You'll see a non-empty sentence explaining ETL beneath the cell. That response confirms ai.generate_text() is available in your hosted runtime.

AI response missing?

  • Confirm the Google Account age is 18+.
  • Confirm the account uses a locale supported by Colab AI.
  • Run the verification cell again after confirming AI access.

Help me troubleshoot why the Gemini footer or ai.generate_text is unavailable in Google Colab.

That's your browser workspace fully checked. Python runs in the notebook.

The Gemini footer responds. The Python AI interface also returns text.

Your hosted notebook is ready for the pipeline. Next up, you'll load a synthetic orders file and watch dirty data break a naive conversion.

Load Messy Orders and Observe Failure

Your Google Colab workspace is ready. A data pipeline still needs raw input that exposes the mess it must handle.

Retail order exports often arrive as CSV text. Every field starts as a string.

One malformed price can stop an unguarded numeric conversion. This step lets you observe that failure before you build the validation rules.

In this step, get ready to:
  • Create a synthetic orders file containing nine messy records.
  • Load the CSV records as dictionaries for inspection.
  • Use Colab AI for quality-check hypotheses before testing a naive conversion.
Create the messy source

Synthetic data gives you realistic quality problems without exposing personal information. This source includes duplicate identifiers plus missing values.

It also includes inconsistent formatting plus a price that cannot become a number. These defects give your pipeline meaningful cases to test.

  • Add a new code cell beneath your temporary AI verification cell.
  • Paste this code into the new cell:
# Cell 1: Imports and synthetic source
import csv
import json
import sqlite3
from google.colab import ai

RAW_CSV = """order_id,order_date,customer_email,product,quantity,unit_price
1001,2026-08-01,ava@example.com,Keyboard,1,45.00
1002,2026-08-01,ben@example.com,Mouse,2,18.50
1002,2026-08-01,ben@example.com,Mouse,2,18.50
1003,2026-08-02,,Webcam,1,62.00
1004,2026-08-02,dee@example.com,USB Cable,0,9.99
1005,2026-08-03,eli@example.com,Laptop Stand,1,not-a-number
1006,2026-08-03,FAY@EXAMPLE.COM,  monitor  ,1,175.00
1007,2026-08-04,gia@example.com,Headset,2,35.50
1008,2026-08-04,hue@example.com,,1,12.00
"""

with open("raw_orders.csv", "w", newline="", encoding="utf-8") as file:
    file.write(RAW_CSV)

What Does This Code Do?

  • The csv module handles the raw orders file.
  • The json import supports later JSON reporting.
  • The sqlite3 import provides access to SQLite.
  • The ai import keeps the verified Colab AI helper available.
  • RAW_CSV holds the header plus nine synthetic orders.
  • The with open(...) block writes that text to raw_orders.csv in the hosted runtime.
  • Run the cell with its run button.
  • Select the file browser icon in the left sidebar.

You should see raw_orders.csv in the runtime file list. The file contains the nine synthetic orders from RAW_CSV.

Cannot See the Source File?

Rerun the source cell if raw_orders.csv is missing. Confirm that the cell finishes without an error.

Check that the filename inside open(...) matches raw_orders.csv exactly.

Help me find why my synthetic CSV file is missing from the Colab runtime.

Ingest and inspect the records

A csv.DictReader uses the header row as dictionary keys. The remaining values stay as strings until your pipeline converts them.

  • Place your cursor beneath file.write(RAW_CSV) in the source cell.
  • Paste this code at the bottom of the same cell:
with open("raw_orders.csv", newline="", encoding="utf-8") as file:
    raw_rows = list(csv.DictReader(file))

print(f"Loaded {len(raw_rows)} raw rows")
print("First raw row:", raw_rows[0])

How Does Ingestion Work?

  • csv.DictReader uses each column name as a dictionary key.
  • list(...) collects all nine dictionaries in memory.
  • raw_rows becomes the list your validation rules can process.
  • The two print(...) calls expose the row count plus one ingested record.
  • Run the complete source cell again.

You should see Loaded 9 raw rows. The next line begins with First raw row:.

That first dictionary shows order 1001. You now have nine raw records available for inspection.

Seeing the Wrong Row Count?

Confirm that RAW_CSV contains the header plus all nine data rows. A missing data line changes the loaded count.

Rerun the complete source cell so raw_orders.csv is rewritten before csv.DictReader reads it.

Help me diagnose why my CSV ingestion does not load nine dictionaries.

Test AI hypotheses against the naive path

AI can suggest useful checks from column names plus sample rows. Those suggestions remain hypotheses until deterministic code tests the records.

  • Switch back to the Colab AI panel from earlier.
  • Ask it which quality checks it proposes for the six column names in RAW_CSV plus the first three synthetic rows.
  • Review its response as a list of candidate checks.

Why Treat AI Suggestions as Hypotheses?

Colab AI can identify likely risks such as duplicates or malformed numbers. Its response does not prove which records are valid.

Deterministic rules produce repeatable decisions from the same input. The observable pipeline results remain your source of truth.

  • Add a temporary code cell beneath the ingestion cell.
  • Pause to predict whether all nine rows can become totals successfully.
  • Test the naive conversion by running this code:
naive_totals = [int(row["quantity"]) * float(row["unit_price"]) for row in raw_rows]

Why Did the Conversion Fail?

The list comprehension attempts to convert every quantity plus every unit price in one pass. Python raises a ValueError when float(...) reaches not-a-number.

The entire conversion stops at that record. This planned failure proves that one malformed field can halt all-or-nothing processing.

  • Leave the failed temporary cell visible beneath the working ingestion cell.

Did the Planned Failure Look Different?

Rerun the complete ingestion cell if the temporary cell cannot find raw_rows. The successful ingestion output must appear before this test.

Confirm that the synthetic price for order 1005 still reads not-a-number.

Help me understand the output from my planned naive conversion failure.

✔️ Awesome, I've got everything!

Your source cell now creates and inspects nine raw orders. The temporary conversion remains separate from the pipeline code.

ⓧ I'd like to double check the full code

Compare your complete source cell with this cumulative notebook code:

# Cell 1: Imports and synthetic source
import csv
import json
import sqlite3
from google.colab import ai

RAW_CSV = """order_id,order_date,customer_email,product,quantity,unit_price
1001,2026-08-01,ava@example.com,Keyboard,1,45.00
1002,2026-08-01,ben@example.com,Mouse,2,18.50
1002,2026-08-01,ben@example.com,Mouse,2,18.50
1003,2026-08-02,,Webcam,1,62.00
1004,2026-08-02,dee@example.com,USB Cable,0,9.99
1005,2026-08-03,eli@example.com,Laptop Stand,1,not-a-number
1006,2026-08-03,FAY@EXAMPLE.COM,  monitor  ,1,175.00
1007,2026-08-04,gia@example.com,Headset,2,35.50
1008,2026-08-04,hue@example.com,,1,12.00
"""

with open("raw_orders.csv", "w", newline="", encoding="utf-8") as file:
    file.write(RAW_CSV)

with open("raw_orders.csv", newline="", encoding="utf-8") as file:
    raw_rows = list(csv.DictReader(file))

print(f"Loaded {len(raw_rows)} raw rows")
print("First raw row:", raw_rows[0])

You have a reproducible raw-data source plus visible proof that naive conversion is fragile. Next, you will replace the all-or-nothing path with rules that quarantine invalid records.

Validate and Quarantine Orders

Your nine raw orders are loaded. The failed conversion has shown how one malformed price can stop the whole pipeline.

Inside Google Colab, deterministic data-quality validation decides each record's path. Valid rows enter clean_rows. Bad rows enter rejected_rows through a quarantine path that records the reason.

In this step, get ready to:
  • Normalize raw values before checking them.
  • Apply deterministic rules that separate valid orders from rejected orders.
  • Compare an AI explanation with the pipeline's observable results.
Prepare the validation state

Normalization removes surrounding spaces before any comparison happens. The validation state also tracks accepted IDs plus every rejection reason.

  • Add a new code cell beneath the temporary cell that raised the conversion error.
  • Paste this setup into the new code cell:
# Cell 2: Normalize, validate, and quarantine
required_fields = [
    "order_id",
    "order_date",
    "customer_email",
    "product",
    "quantity",
    "unit_price",
]

clean_rows = []
rejected_rows = []
seen_order_ids = set()
reason_counts = {}


def reject(row, reason):
    rejected = dict(row)
    rejected["rejection_reason"] = reason
    rejected_rows.append(rejected)
    reason_counts[reason] = reason_counts.get(reason, 0) + 1

What does this setup do?

  • required_fields lists the values that every usable order needs.
  • clean_rows holds records that pass every rule.
  • rejected_rows holds quarantined records with an added reason.
  • reject() copies a bad row before recording its reason in reason_counts.
  • Run the code cell with the play button.

The cell should finish without an exception. Your validation containers are now ready for the raw records.

Does the setup cell fail?

Confirm that every opening bracket has a matching closing bracket. Check that the body of reject() stays indented beneath the function definition.

Help me fix my validation setup cell.

Apply the deterministic rules

The first rules catch missing fields plus invalid quantities. Each rejected row exits its current loop pass through continue so later calculations cannot use it.

  • Place your cursor below the reject() function in the same code cell.
  • Paste this first set of rules:
for row in raw_rows:
    normalized = {
        key: (value or "").strip()
        for key, value in row.items()
    }

    if any(not normalized[field] for field in required_fields):
        reject(normalized, "missing_required_value")
        continue

    if normalized["order_id"] in seen_order_ids:
        reject(normalized, "duplicate_order_id")
        continue

    if normalized["customer_email"].count("@") != 1:
        reject(normalized, "invalid_email")
        continue

    try:
        quantity = int(normalized["quantity"])
    except ValueError:
        reject(normalized, "quantity_not_integer")
        continue

    if quantity <= 0:
        reject(normalized, "quantity_must_be_positive")
        continue

print(f"Clean rows: {len(clean_rows)} | Rejected rows: {len(rejected_rows)}")
print("Rejection counts:", reason_counts)

How do these first rules work?

  • strip() removes surrounding whitespace from every raw value.
  • any() quarantines a row when at least one required value is empty.
  • count("@") requires each customer email to contain exactly one at sign.
  • try protects the integer conversion. except ValueError gives malformed quantities a rejection reason instead of stopping the loop.

Before you run this checkpoint, how many rows do you expect these first rules to quarantine?

  • Run the updated code cell with the play button.

You'll see Clean rows: 0 | Rejected rows: 3. This checkpoint has caught two missing required values plus one nonpositive quantity.

Seeing a different checkpoint count?

Confirm that the new loop remains in the same cell as the empty validation containers. Running the whole cell resets those containers before checking all nine rows.

Check the indentation beneath each condition. Every matching reject() call needs its following continue at the same indentation level.

Help me find why my checkpoint count differs.

The remaining rules validate prices before building typed clean records. Accepted IDs are also recorded here so later duplicates enter quarantine.

  • Replace the two temporary print lines at the bottom of the code cell with the remaining rules below:
    try:
        unit_price = float(normalized["unit_price"])
    except ValueError:
        reject(normalized, "unit_price_not_numeric")
        continue

    if unit_price <= 0:
        reject(normalized, "unit_price_must_be_positive")
        continue

    seen_order_ids.add(normalized["order_id"])
    clean_rows.append(
        {
            "order_id": normalized["order_id"],
            "order_date": normalized["order_date"],
            "customer_email": normalized["customer_email"].lower(),
            "product": normalized["product"].title(),
            "quantity": quantity,
            "unit_price": unit_price,
            "order_total": round(quantity * unit_price, 2),
        }
    )

print(f"Clean rows: {len(clean_rows)} | Rejected rows: {len(rejected_rows)}")
print("Rejection counts:", reason_counts)

What completes the validation path?

  • float() converts a valid price string into a number.
  • except ValueError quarantines a nonnumeric price without ending the pipeline.
  • seen_order_ids records accepted order IDs so a later copy can be rejected.
  • clean_rows.append() lowercases the email. It title-cases the product. It calculates the order total only after every rule passes.

Before you rerun the cell, do you think the malformed price will crash the pipeline again?

  • Run the completed validation cell with the play button.

You'll see Clean rows: 4 | Rejected rows: 5. The rejection counts show one duplicate order ID plus two missing values.

They also show one nonpositive quantity plus one nonnumeric price. That's the core safeguard working: one bad price no longer blocks the four valid orders.

Not seeing four clean rows?

Confirm that seen_order_ids.add() appears before clean_rows.append(). This placement lets the second occurrence of an accepted order ID trigger the duplicate rule.

Check that the price conversion sits inside its own try block. The matching exception path must call reject() before continuing.

Help me debug my clean and rejected counts.

Review the guardrails with AI

The printed counts are evidence produced by code. Colab AI can explain the control flow, but its explanation remains an interpretation.

  • Return to the Colab AI chat panel from earlier.
  • Ask why the try blocks keep malformed numeric values from stopping the validation loop.
  • Ask how the matching except ValueError blocks send those records into quarantine.
  • Compare the explanation with the printed clean count.
  • Compare the explanation with the printed rejection reasons.

A useful explanation matches the behavior you observed. The deterministic output remains the source of truth.

  • Delete the temporary naive_totals code cell now that the guarded validation has replaced it.

Your notebook should now contain the synthetic source cell followed by the guarded validation cell. The successful four-to-five split remains visible beneath the validation cell.

✔️ Awesome, I've got everything!

Great. Double check that the validation cell is saved in your AI Data Quality Pipeline notebook.

ⓧ I'd like to double check the full code

Compare your two notebook cells with the complete code below.

# Cell 1: Imports and synthetic source
import csv
import json
import sqlite3
from google.colab import ai

RAW_CSV = """order_id,order_date,customer_email,product,quantity,unit_price
1001,2026-08-01,ava@example.com,Keyboard,1,45.00
1002,2026-08-01,ben@example.com,Mouse,2,18.50
1002,2026-08-01,ben@example.com,Mouse,2,18.50
1003,2026-08-02,,Webcam,1,62.00
1004,2026-08-02,dee@example.com,USB Cable,0,9.99
1005,2026-08-03,eli@example.com,Laptop Stand,1,not-a-number
1006,2026-08-03,FAY@EXAMPLE.COM,  monitor  ,1,175.00
1007,2026-08-04,gia@example.com,Headset,2,35.50
1008,2026-08-04,hue@example.com,,1,12.00
"""

with open("raw_orders.csv", "w", newline="", encoding="utf-8") as file:
    file.write(RAW_CSV)

with open("raw_orders.csv", newline="", encoding="utf-8") as file:
    raw_rows = list(csv.DictReader(file))

print(f"Loaded {len(raw_rows)} raw rows")
print("First raw row:", raw_rows[0])

# Cell 2: Normalize, validate, and quarantine
required_fields = [
    "order_id",
    "order_date",
    "customer_email",
    "product",
    "quantity",
    "unit_price",
]

clean_rows = []
rejected_rows = []
seen_order_ids = set()
reason_counts = {}


def reject(row, reason):
    rejected = dict(row)
    rejected["rejection_reason"] = reason
    rejected_rows.append(rejected)
    reason_counts[reason] = reason_counts.get(reason, 0) + 1


for row in raw_rows:
    normalized = {
        key: (value or "").strip()
        for key, value in row.items()
    }

    if any(not normalized[field] for field in required_fields):
        reject(normalized, "missing_required_value")
        continue

    if normalized["order_id"] in seen_order_ids:
        reject(normalized, "duplicate_order_id")
        continue

    if normalized["customer_email"].count("@") != 1:
        reject(normalized, "invalid_email")
        continue

    try:
        quantity = int(normalized["quantity"])
    except ValueError:
        reject(normalized, "quantity_not_integer")
        continue

    if quantity <= 0:
        reject(normalized, "quantity_must_be_positive")
        continue

    try:
        unit_price = float(normalized["unit_price"])
    except ValueError:
        reject(normalized, "unit_price_not_numeric")
        continue

    if unit_price <= 0:
        reject(normalized, "unit_price_must_be_positive")
        continue

    seen_order_ids.add(normalized["order_id"])
    clean_rows.append(
        {
            "order_id": normalized["order_id"],
            "order_date": normalized["order_date"],
            "customer_email": normalized["customer_email"].lower(),
            "product": normalized["product"].title(),
            "quantity": quantity,
            "unit_price": unit_price,
            "order_total": round(quantity * unit_price, 2),
        }
    )

print(f"Clean rows: {len(clean_rows)} | Rejected rows: {len(rejected_rows)}")
print("Rejection counts:", reason_counts)

Your pipeline now preserves four trusted orders while quarantining five bad records with clear reasons. Next up, you'll publish these results as auditable CSV and JSON artifacts.

Publish Quality Outputs and Briefing

Your validation rules now separate four trusted orders from five quarantined records. Those results still live inside your Google Colab notebook memory.

Auditable outputs make those decisions easy to inspect. In this step, you'll publish two CSV files plus a deterministic JSON report. Colab AI will brief you using only those numbers.

In this step, get ready to:
  • Publish trusted records to clean_orders.csv.
  • Publish quarantined records to rejected_orders.csv.
  • Generate a deterministic quality report that Colab AI summarizes.
Prepare auditable CSV files

Separate files give trusted records a clear destination. They also preserve rejected records with the reason each one failed.

  • Switch back to the Google Colab notebook from earlier.
  • Add a code cell below the validation cell by using the code-cell control in the notebook toolbar.
  • Paste the CSV publishing code into the new cell:
# Cell 3: Publish CSV and JSON quality artifacts
clean_fields = [
    "order_id",
    "order_date",
    "customer_email",
    "product",
    "quantity",
    "unit_price",
    "order_total",
]

with open("clean_orders.csv", "w", newline="", encoding="utf-8") as file:
    writer = csv.DictWriter(file, fieldnames=clean_fields)
    writer.writeheader()
    writer.writerows(clean_rows)

rejected_fields = required_fields + ["rejection_reason"]
with open("rejected_orders.csv", "w", newline="", encoding="utf-8") as file:
    writer = csv.DictWriter(file, fieldnames=rejected_fields)
    writer.writeheader()
    writer.writerows(rejected_rows)

What does this code publish?

  • The clean_fields list defines the output schema for trusted records.
  • The two csv.DictWriter objects write headers followed by their matching rows.
  • The rejected_fields list adds rejection_reason to the original source fields.
  • Run the code cell by clicking its play button.
  • Select the folder icon in the left sidebar to inspect the runtime files.

You'll see clean_orders.csv beside rejected_orders.csv in the runtime file list.

Good progress. Your trusted records now have a separate audit trail from the quarantined records.

Files missing from the runtime?

Rerun the earlier validation cell if the runtime no longer contains clean_rows or rejected_rows. Then rerun the publishing cell.

Check that both filenames end with .csv.

Help me troubleshoot missing CSV outputs in my Colab notebook.

Generate the deterministic quality report

A quality report turns the pipeline's actual results into metrics you can audit. Its values come directly from the lists created by your validation rules.

  • Place your cursor on a blank line at the end of the publishing cell.
  • Extend the cell by pasting this code:
report = {
    "source_file": "raw_orders.csv",
    "input_rows": len(raw_rows),
    "valid_rows": len(clean_rows),
    "rejected_rows": len(rejected_rows),
    "valid_percentage": round(len(clean_rows) / len(raw_rows) * 100, 1),
    "rejection_reasons": reason_counts,
}

with open("quality_report.json", "w", encoding="utf-8") as file:
    json.dump(report, file, indent=2)

print(json.dumps(report, indent=2))

How is the report calculated?

  • The report dictionary derives every count from raw_rows or the validation results.
  • The valid_percentage calculation measures how much of the source passed every rule.
  • The json.dump call saves the report to quality_report.json.
  • The json.dumps call formats the same report for the notebook output.
  • Run the entire publishing cell again.

You'll see input_rows as 9. You'll see valid_rows as 4.

You'll see rejected_rows as 5. You'll see valid_percentage as 44.4.

That's an auditable checkpoint. The report reflects the records your code actually accepted or rejected.

Report values look different?

Rerun the source cell followed by the validation cell. This resets the in-memory lists before the report calculates its metrics.

Check that valid_percentage uses the lengths of clean_rows and raw_rows.

Help me debug the values in my deterministic quality report.

Create the AI quality briefing

The deterministic report remains the source of truth. The AI briefing adds a readable interpretation without deciding which records are valid.

  • Place your cursor on a blank line at the end of the same publishing cell.
  • Add the constrained AI briefing by pasting this code:
ai_prompt = (
    "You are helping a beginner data engineer. Summarize this deterministic "
    "data-quality report in three short bullet points. Do not invent facts.\n"
    + json.dumps(report, indent=2)
)
ai_summary = ai.generate_text(ai_prompt)
print("AI quality briefing:")
print(ai_summary)

What keeps the briefing grounded?

  • The ai_prompt includes only the deterministic report built from synthetic data.
  • The prompt requests three short bullets based on existing facts.
  • The ai.generate_text(ai_prompt) call produces the interpretation stored in ai_summary.
  • The AI quality briefing: label keeps the generated interpretation distinct from the report.

Before you rerun the cell, do you expect the report numbers or the briefing wording to stay fixed?

  • Run the entire publishing cell one final time.

You'll first see the deterministic report with 9 input rows. It shows 4 valid rows.

The report also shows 5 rejected rows. Its valid percentage is 44.4.

Below the report, you'll see AI quality briefing: followed by a non-empty three-bullet response.

Strong finish. Your AI summary can change its wording while the validated metrics remain fixed.

Briefing missing or empty?

Confirm that your Colab AI access still works by rerunning the AI verification cell from earlier. Keep the synthetic report as the only data included in the prompt.

Check that report exists before ai.generate_text(ai_prompt) runs.

Help me troubleshoot an empty Colab AI quality briefing.

✔️ Awesome, I've got everything!

Your publishing cell now creates both CSV outputs. It also writes the JSON report before generating the AI briefing.

ⓧ I'd like to double check the full code

  • Compare your publishing cell with this complete version:
# Cell 3: Publish CSV and JSON quality artifacts
clean_fields = [
    "order_id",
    "order_date",
    "customer_email",
    "product",
    "quantity",
    "unit_price",
    "order_total",
]

with open("clean_orders.csv", "w", newline="", encoding="utf-8") as file:
    writer = csv.DictWriter(file, fieldnames=clean_fields)
    writer.writeheader()
    writer.writerows(clean_rows)

rejected_fields = required_fields + ["rejection_reason"]
with open("rejected_orders.csv", "w", newline="", encoding="utf-8") as file:
    writer = csv.DictWriter(file, fieldnames=rejected_fields)
    writer.writeheader()
    writer.writerows(rejected_rows)

report = {
    "source_file": "raw_orders.csv",
    "input_rows": len(raw_rows),
    "valid_rows": len(clean_rows),
    "rejected_rows": len(rejected_rows),
    "valid_percentage": round(len(clean_rows) / len(raw_rows) * 100, 1),
    "rejection_reasons": reason_counts,
}

with open("quality_report.json", "w", encoding="utf-8") as file:
    json.dump(report, file, indent=2)

print(json.dumps(report, indent=2))

ai_prompt = (
    "You are helping a beginner data engineer. Summarize this deterministic "
    "data-quality report in three short bullet points. Do not invent facts.\n"
    + json.dumps(report, indent=2)
)
ai_summary = ai.generate_text(ai_prompt)
print("AI quality briefing:")
print(ai_summary)

Your pipeline now publishes evidence that others can inspect. Next, you'll load only the trusted rows into SQLite. A query will prove what reached the warehouse.

Load and Query the SQLite Warehouse

Your clean and rejected files now prove which orders passed the quality rules. The final core step turns those trusted records into something you can query.

Flat files are useful audit artifacts. An SQLite database completes the data-engineering flow by loading typed records into a table.

In this step, get ready to:
  • Create a typed warehouse table for validated orders.
  • Load the four clean records with parameter placeholders.
  • Query the warehouse for its row count plus total revenue.
Create the warehouse table

A table schema defines which fields the warehouse accepts. The primary key also prevents two records from sharing the same order ID.

  • Add a new code cell directly below Cell 3.
  • Build the table in the new cell with this code:
# Cell 4: Load and query SQLite
connection = sqlite3.connect("orders.db")
connection.execute(
    """
    CREATE TABLE IF NOT EXISTS clean_orders (
        order_id TEXT PRIMARY KEY,
        order_date TEXT NOT NULL,
        customer_email TEXT NOT NULL,
        product TEXT NOT NULL,
        quantity INTEGER NOT NULL,
        unit_price REAL NOT NULL,
        order_total REAL NOT NULL
    )
    """
)
connection.execute("DELETE FROM clean_orders")

What does this code do?

  • The sqlite3.connect() call creates orders.db or connects to the existing database.
  • The CREATE TABLE statement gives each clean field a database type plus a required-value rule.
  • The primary key makes every order_id unique inside the table.
  • The DELETE FROM clean_orders statement clears earlier rows before each load.
  • Run Cell 4.
  • Check the notebook file browser for orders.db.

You should see orders.db beside your CSV and JSON artifacts. The database now contains the empty clean_orders table.

Database file missing?

  • Confirm that Cell 4 begins with # Cell 4: Load and query SQLite.
  • Rerun the imports cell from earlier if Python does not recognize sqlite3.
  • Refresh the notebook file browser after Cell 4 finishes.

Ask for help with the exact cell output: Help me find why orders.db was not created in my Google Colab runtime.

Load the clean records safely

Parameter placeholders keep each Python value separate from the SQL statement. This gives the database a consistent way to bind every clean field.

  • Place your cursor after connection.execute("DELETE FROM clean_orders") in Cell 4.
  • Append the safe load logic with this code:
connection.executemany(
    """
    INSERT INTO clean_orders (
        order_id,
        order_date,
        customer_email,
        product,
        quantity,
        unit_price,
        order_total
    ) VALUES (?, ?, ?, ?, ?, ?, ?)
    """,
    [
        (
            row["order_id"],
            row["order_date"],
            row["customer_email"],
            row["product"],
            row["quantity"],
            row["unit_price"],
            row["order_total"],
        )
        for row in clean_rows
    ],
)
connection.commit()

How Does the Safe Load Work?

  • The executemany() method runs the same insert for every record in clean_rows.
  • Each ? placeholder receives one value from an order tuple.
  • The field order in each tuple matches the column order in the insert statement.
  • The commit() call persists the four inserted records.
  • Run Cell 4 again.

You should see the cell finish without an exception. That successful run confirms the table accepted all four bound records.

Clean Rows Not Loading?

  • Rerun the earlier validation cell if Python does not recognize clean_rows.
  • Confirm that the seven tuple values match the seven insert columns.
  • Keep the load code above the query code you add next.

Share the exception without sharing sensitive data: Help me debug why executemany cannot load my clean_rows into clean_orders.

Query and verify the warehouse

A SQL query gives you a database-level view of the records that were loaded. The result lets you reconcile the row count plus revenue with the clean pipeline output.

  • Place your cursor after connection.commit() in Cell 4.
  • Append the warehouse query plus summary with this code:
warehouse_rows = connection.execute(
    "SELECT order_id, order_total FROM clean_orders ORDER BY order_id"
).fetchall()
connection.close()

warehouse_revenue = round(sum(row[1] for row in warehouse_rows), 2)
print(
    f"Warehouse rows: {len(warehouse_rows)} | "
    f"Warehouse revenue: ${warehouse_revenue:.2f}"
)

How Does the Query Verify the Load?

  • The query selects each order ID plus its calculated total.
  • The ORDER BY order_id clause returns the records in a predictable order.
  • The fetchall() method stores the query result as tuples in warehouse_rows.
  • The revenue calculation adds the second value from every returned tuple.
  • The close() call releases the database connection after the query.
  • Open Colab AI by clicking the Gemini spark icon in the notebook footer.
  • Ask Colab AI to explain how the SQL query selects plus orders the warehouse results in plain English.

Treat the AI response as an explanation. Cell 4 remains the verification source.

Before you run Cell 4, predict how many rows the warehouse query returns.

  • Run Cell 4 from top to bottom.

You should see Warehouse rows: 4 | Warehouse revenue: $328.00 in the output.

That result closes the loop. Your four trusted orders are now queryable in SQLite.

Warehouse Totals Do Not Match?

  • Rerun the validation cell to restore the four records in clean_rows.
  • Confirm that Cell 4 deletes earlier table rows before inserting the current records.
  • Confirm that the final query reads from clean_orders.

Use the observed count plus revenue in your request: Help me reconcile my SQLite warehouse result with the expected four rows and $328.00.

✔️ Awesome, I've got everything!

Your completed Cell 4 creates the table, reloads the trusted records, queries the result, and closes the connection.

ⓧ I'd like to double check the full code

Compare Cell 4 with this complete version:

# Cell 4: Load and query SQLite
connection = sqlite3.connect("orders.db")
connection.execute(
    """
    CREATE TABLE IF NOT EXISTS clean_orders (
        order_id TEXT PRIMARY KEY,
        order_date TEXT NOT NULL,
        customer_email TEXT NOT NULL,
        product TEXT NOT NULL,
        quantity INTEGER NOT NULL,
        unit_price REAL NOT NULL,
        order_total REAL NOT NULL
    )
    """
)
connection.execute("DELETE FROM clean_orders")
connection.executemany(
    """
    INSERT INTO clean_orders (
        order_id,
        order_date,
        customer_email,
        product,
        quantity,
        unit_price,
        order_total
    ) VALUES (?, ?, ?, ?, ?, ?, ?)
    """,
    [
        (
            row["order_id"],
            row["order_date"],
            row["customer_email"],
            row["product"],
            row["quantity"],
            row["unit_price"],
            row["order_total"],
        )
        for row in clean_rows
    ],
)
connection.commit()
warehouse_rows = connection.execute(
    "SELECT order_id, order_total FROM clean_orders ORDER BY order_id"
).fetchall()
connection.close()

warehouse_revenue = round(sum(row[1] for row in warehouse_rows), 2)
print(
    f"Warehouse rows: {len(warehouse_rows)} | "
    f"Warehouse revenue: ${warehouse_revenue:.2f}"
)

Secret mission

Secret Mission: Add a Cross-Layer Reconciliation Gate

Your warehouse query returned the expected row count and revenue. This challenge adds assertions that prove the warehouse contains the same order identities and value as the clean pipeline output. You will also trigger a controlled mismatch to confirm the gate fails when the two layers drift apart.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep your work available, pause your notebook session, or delete the project resources. Your Google Colab workspace uses free-of-charge resources with no ongoing project cost.

Resources you used:

  • The AI Data Quality Pipeline.ipynb notebook stored in Google Drive.
  • The raw CSV source stored as raw_orders.csv in the temporary Colab virtual machine.
  • The trusted output stored as clean_orders.csv in the temporary Colab virtual machine.
  • The quarantined output stored as rejected_orders.csv in the temporary Colab virtual machine.
  • The JSON quality report stored as quality_report.json in the temporary Colab virtual machine.
  • The SQLite warehouse stored as orders.db in the temporary Colab virtual machine.

Keep everything running

No action is required. Choose this option while you are reviewing the pipeline or extending its quality checks.

  • Keep AI Data Quality Pipeline.ipynb in Google Drive.
  • Download each runtime artifact listed above that you want to retain.

The notebook stays in Google Drive. Runtime files can disappear when the virtual machine is deleted after idling.

Pause - I'll come back to this later

Close the current session while keeping the notebook ready for another day.

  • Download each runtime artifact that you want to retain.
  • Close the Google Colab browser tab.
  • Reopen AI Data Quality Pipeline.ipynb from Google Drive when you are ready to continue.

Your notebook remains in Google Drive. The temporary runtime files may disappear while the notebook is inactive.

Delete - I don't want to use this again

Remove the saved notebook from Google Drive. Reset the temporary Colab virtual machine to clear its generated artifacts.

Save Files Before Deletion

Runtime deletion permanently removes files inside the temporary virtual machine. Copies downloaded to your device remain available.

  • Download every runtime artifact that you want to preserve.
  • Select Runtime from the notebook menu.
  • Select Disconnect and delete runtime.
  • Confirm the deletion when Colab asks.

You should see that the notebook is disconnected from its runtime. This clears raw_orders.csv, clean_orders.csv, rejected_orders.csv, quality_report.json, and orders.db from the temporary virtual machine.

  • Open Google Drive in a new browser tab.
  • Locate AI Data Quality Pipeline.ipynb in the folder where Colab saved it.
  • Use the notebook file's context menu to delete it.
  • Confirm AI Data Quality Pipeline.ipynb no longer appears in that Google Drive folder.

Cleanup complete. Your Colab runtime is reset.

Nice Work!

Nice Work!

What a finish! Your Google Colab notebook now turns nine messy retail orders into four trusted records.

You've learned how to:

  • Build a repeatable ETL pipeline that ingests nine synthetic orders from a CSV file. Catch malformed values with deterministic rules after observing the naive conversion fail on not-a-number. Produce four normalized records. Quarantine five invalid records with explicit reasons.
  • Publish clean_orders.csv, rejected_orders.csv, and quality_report.json as audit-ready artifacts. Use Colab AI to summarize deterministic JSON metrics without giving the model authority over validation.
  • Load four validated records into a typed SQLite table. Run a SQL query that proves the warehouse contains four orders worth $328.00.
  • Secret Mission: Add a cross-layer reconciliation gate that compares warehouse order IDs with clean_rows. Compare revenue across both layers. Trigger an assertion failure before restoring the passing gate.

Ready to quiz yourself?