Build a CISA KEV Pipeline with DuckDB

Build a local pipeline that turns CISA KEV data into validated Parquet.

Introduction

30 Second Summary

A public vulnerability list changes as new threats are confirmed. Turning that moving list into dependable priorities takes more than downloading a file.

In this project, you will build a local batch ETL pipeline that turns the live CISA Known Exploited Vulnerabilities Catalog into a trustworthy analytical dataset. Python and DuckDB will expose the nested JSON structure before publishing validated Parquet output.

What You'll Build

With one command, you will watch the live catalog become a validated Parquet dataset that is ready for downstream analysis.

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

  • Refresh the source with one command that saves the latest public catalog under data/raw.
  • Diagnose nested data by seeing DuckDB report one top-level row before flattening reveals every vulnerability.
  • Prove data quality with visible PASS results for row counts, required fields, unique CVE IDs, and the exported file.
  • Secret Mission: Publish a due-date-sorted Parquet queue containing only vulnerabilities linked to known ransomware campaigns.

Are there any prerequisites?

You'll need internet access on a Windows computer with Python and Visual Studio Code already installed. Step 1 confirms Python 3.10 or later before installing DuckDB 1.5.5.

Before We Start

Before any hands-on work begins, lock in what you are building and why it matters. Your goal is a Windows-local Python batch pipeline that uses DuckDB to turn nested CISA KEV JSON into a validated analytics dataset for security teams.

Set Up the Local Pipeline Environment

Your batch pipeline depends on a stable local environment. An isolated setup prevents unrelated packages on your computer from changing how the pipeline behaves.

You will prepare the project in Visual Studio Code. You will confirm that Python meets the minimum version before installing DuckDB inside a project-specific virtual environment.

In this step, get ready to:
  • Open the cisa-kev-pipeline folder in Visual Studio Code.
  • Confirm that your computer has Python 3.10 or later.
  • Create an isolated environment with DuckDB 1.5.5 installed.
Open the project folder and check Python

A Visual Studio Code workspace keeps your files beside a terminal that starts in the same folder. This gives every command a predictable location.

  • Press the Windows key to open the search bar.
  • Type Visual Studio Code and press Enter.
  • Select File from the top menu.
  • Select Open Folder....
  • Select your Desktop in the folder picker.
  • Select New Folder.
  • Type cisa-kev-pipeline and press Enter.

Why open a folder as a workspace?

Visual Studio Code treats an opened folder as a workspace. Its integrated terminal starts inside that folder automatically.

A Workspace Trust prompt may appear when you open the folder. This empty workspace contains only the files you create during this project.

  • Click Select Folder.
  • Select Yes, I trust the authors if the Workspace Trust prompt appears.

You will see cisa-kev-pipeline at the top of the Explorer sidebar.

  • Select View from the top menu.
  • Select Terminal.
  • Check the installed Python version by running this command:
python --version

What does this command check?

The command asks the Python installation available to your terminal for its version. DuckDB 1.5.5 requires Python 3.10 or later.

✔️ I see Python 3.10 or later

Your Python version is compatible with DuckDB 1.5.5. You can continue with the environment setup.

ⓧ I see an older version

Your current Python installation is below the minimum version required by this project. Update it before creating the virtual environment.

  • Close Visual Studio Code.
  • Visit the official Python Releases for Windows page.
  • Follow the official installation flow for the latest supported Python release.
  • Reopen Visual Studio Code through Windows search.
  • Open the existing cisa-kev-pipeline folder by repeating the folder-opening steps above.
  • Open the integrated terminal by repeating the terminal steps above.
  • Repeat the version check until the output reports Python 3.10 or later.

ⓧ Command not found

Windows cannot currently find a Python installation from this terminal. Install a supported release before continuing.

  • Close Visual Studio Code.
  • Visit the official Python Releases for Windows page.
  • Follow the official installation flow for the latest supported Python release.
  • Reopen Visual Studio Code through Windows search.
  • Open the existing cisa-kev-pipeline folder by repeating the folder-opening steps above.
  • Open the integrated terminal by repeating the terminal steps above.
  • Repeat the version check until the output reports Python 3.10 or later.
Pin DuckDB and create the environment

The requirements.txt file records the exact package release used by the pipeline. This pin keeps future installations consistent.

Why DuckDB for this pipeline?

DuckDB runs inside the Python process. It persists the warehouse in one local file without a separate database server.

Its JSON and Parquet support keeps extraction, transformation, and export inside one local workflow.

  • Select the file-shaped control with a plus sign at the top of the Explorer sidebar.
  • Type requirements.txt and press Enter.
  • Add the pinned dependency by copying this line into requirements.txt:
duckdb==1.5.5

What does this dependency pin do?

  • The package name identifies DuckDB's Python package.
  • The double equals sign requires the exact 1.5.5 release.
  • The requirements file gives the environment a repeatable dependency list.
  • Save requirements.txt.
  • Confirm that requirements.txt appears in the Explorer sidebar.

Can't see the requirements file?

Check that you created the file inside the cisa-kev-pipeline workspace. Confirm that its name ends with .txt.

Help me find or create requirements.txt.

✔️ Awesome, I've got everything!

Your requirements file now pins the exact DuckDB release used by this project. Make sure the file is saved before continuing.

ⓧ I'd like to double check the full code

duckdb==1.5.5

The virtual environment creates a private package location inside cisa-kev-pipeline. The pipeline can use this environment without depending on packages installed elsewhere on your computer.

  • Switch back to the integrated terminal from earlier.
  • Create the .venv environment by running this command:
python -m venv .venv

What does this command create?

  • Python's built-in environment module creates the isolated .venv folder.
  • Windows stores the environment's Python executable inside its Scripts directory.
  • The command leaves your system-wide packages unchanged.
  • Wait for the terminal prompt to return.
  • Confirm that .venv appears in the Explorer sidebar.

Virtual environment missing?

Check that the terminal prompt points to the cisa-kev-pipeline folder. Repeat the command after resolving any Python error shown in the terminal.

Help me diagnose my .venv creation problem.

Install and verify DuckDB

The environment now has its own Python executable. Using that executable directly avoids activation-policy issues in Windows PowerShell.

  • Install the dependency listed in requirements.txt by running this command:
.\.venv\Scripts\python.exe -m pip install -r requirements.txt

What does this install command do?

  • The path selects the Python executable inside .venv.
  • The package installer reads the dependency pin from requirements.txt.
  • DuckDB 1.5.5 is installed only inside this project's environment.

The terminal prompt returns when the installation finishes.

Dependency installation failed?

Confirm that your internet connection is available. Check that requirements.txt contains the exact dependency pin from the full-code tab.

Help me diagnose the DuckDB installation.

Before you run the final check, which DuckDB version do you expect the isolated environment to report?

  • Verify the installed DuckDB version by running this command:
.\.venv\Scripts\python.exe -c "import duckdb; print(duckdb.__version__)"

What does this verification prove?

  • The command launches the Python executable stored inside .venv.
  • The import confirms that the DuckDB package is available to that executable.
  • The printed value confirms which installed package release the pipeline uses.

You will see 1.5.5 in the terminal. That confirms your isolated pipeline environment is ready to run DuckDB code.

Version missing or incorrect?

Confirm that the verification command starts with the .venv Python path. Check that the dependency installation completed without an error.

Help me fix the DuckDB version check.

Your project now has a pinned environment that reports DuckDB 1.5.5. Next, you will use it to download the live CISA feed and reveal its nested-data trap.

Extract the CISA Feed

Your isolated Python environment is ready for its first live data job. The next goal is to pull the current CISA Known Exploited Vulnerabilities Catalog into your project.

A raw JSON response can load successfully while its useful records remain inside a nested list. You will use DuckDB to expose that hidden structure before transforming it in the next step.

In this step, get ready to:
  • Define the live feed source with reusable path constants.
  • Download the catalog with a checked record count.
  • Compare the vulnerability count with DuckDB's top-level row count.
Define the extraction foundation

The script needs a source URL. Path constants keep each later read or write pointed at the same raw file.

  • Select the file creation icon beside the cisa-kev-pipeline folder in the Explorer sidebar.
  • Enter extract.py as the file name.
  • Add the imports and path constants by pasting this code:
import json
import urllib.request
from pathlib import import Path

import duckdb

CISA_KEV_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
RAW_DIR = Path("data/raw")
RAW_FILE = RAW_DIR / "known_exploited_vulnerabilities.json"

What does this foundation do?

  • The json module turns the downloaded bytes into Python data.
  • The urllib.request module sends the web request without another dependency.
  • Path builds folder locations that work on Windows.
  • CISA_KEV_URL stores the official catalog endpoint.
  • RAW_FILE identifies the raw catalog file that the pipeline creates.
  • Save extract.py.
  • Confirm that the Explorer sidebar lists extract.py inside cisa-kev-pipeline.

Is DuckDB underlined?

Visual Studio Code may be checking the file with a different Python interpreter. Choose the interpreter inside .venv for this project.

Help me connect Visual Studio Code to this project's virtual environment.

Download and validate the live catalog

An HTTP request returns the live catalog as bytes. The feed's own record count provides an immediate check against its nested vulnerability list.

  • Place the cursor below the RAW_FILE definition in extract.py.
  • Add the download and validation function by pasting this code:
def extract():
    RAW_DIR.mkdir(parents=True, exist_ok=True)

    request = urllib.request.Request(
        CISA_KEV_URL,
        headers={"User-Agent": "cisa-kev-pipeline/1.0"},
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        raw_bytes = response.read()

    payload = json.loads(raw_bytes)
    expected_count = payload["count"]
    vulnerabilities = payload["vulnerabilities"]

    if expected_count != len(vulnerabilities):
        raise ValueError(
            "CISA feed count does not match the vulnerabilities list length."
        )

    RAW_FILE.write_bytes(raw_bytes)


if __name__ == "__main__":
    extract()

How does extraction stay trustworthy?

  • RAW_DIR.mkdir() creates the raw data folders when they are missing.
  • urllib.request.Request() prepares the request for the CISA feed.
  • json.loads() parses the response so the script can inspect its fields.
  • expected_count must match the length of vulnerabilities before the raw file is saved.
  • RAW_FILE.write_bytes() preserves the downloaded response for later transformations.
  • The final guard calls extract() when you run the file directly.
  • Save extract.py.
  • Show the terminal panel at the bottom of Visual Studio Code.
  • Download the current catalog by running this command:
.\.venv\Scripts\python.exe extract.py

What should happen?

The command runs extract.py with the Python interpreter from .venv. The script downloads the current feed into data/raw/known_exploited_vulnerabilities.json.

You should see the new raw file in the Explorer sidebar. That first batch extract is complete: every later transformation now starts from the saved source bytes.

Is the raw file missing?

Confirm that your computer can reach the internet. Check that the terminal is running from the cisa-kev-pipeline folder.

Help me debug why extract.py did not save the CISA JSON file.

Inspect the nested payload with DuckDB

DuckDB can read the saved JSON directly. Counting its top-level rows reveals the shape that an analytical query receives before any flattening happens.

  • Find RAW_FILE.write_bytes(raw_bytes) inside extract().
  • Place the cursor on the blank line below it.
  • Add the DuckDB inspection and printed report by pasting this code:
    with duckdb.connect() as connection:
        top_level_rows = (
            connection.read_json(str(RAW_FILE)).count("*").fetchone()[0]
        )

    print(f"CISA catalog version: {payload['catalogVersion']}")
    print(f"CISA vulnerability records: {expected_count}")
    print(f"DuckDB top-level rows before flattening: {top_level_rows}")
    return expected_count

What does the inspection measure?

  • duckdb.connect() creates a temporary in-memory connection for this inspection.
  • read_json() infers the saved response's top-level structure.
  • count("*") measures the number of rows in that inferred structure.
  • The printed values place the live vulnerability count beside DuckDB's top-level count.
  • Save extract.py.

Before you run the completed script, how many top-level rows do you think DuckDB will report for a catalog containing many vulnerabilities?

  • Test the completed extraction by running this command:
.\.venv\Scripts\python.exe extract.py

What should you see?

The catalog version reflects the live feed at the time you run the script. The CISA vulnerability count is greater than one.

The final line reads DuckDB top-level rows before flattening: 1. DuckDB sees the outer JSON object as one row because the vulnerability records remain inside its nested list.

Strong result. You have exposed the exact nested-data trap that can derail a batch pipeline.

Did the inspection fail?

Check that data/raw/known_exploited_vulnerabilities.json exists. Compare the indentation inside extract() with the full file below.

Help me debug the DuckDB inspection in extract.py.

✔️ Awesome, I've got everything!

Great. Save extract.py before moving on.

ⓧ I'd like to double check the full code

import json
import urllib.request
from pathlib import Path

import duckdb

CISA_KEV_URL = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
RAW_DIR = Path("data/raw")
RAW_FILE = RAW_DIR / "known_exploited_vulnerabilities.json"


def extract():
    RAW_DIR.mkdir(parents=True, exist_ok=True)

    request = urllib.request.Request(
        CISA_KEV_URL,
        headers={"User-Agent": "cisa-kev-pipeline/1.0"},
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        raw_bytes = response.read()

    payload = json.loads(raw_bytes)
    expected_count = payload["count"]
    vulnerabilities = payload["vulnerabilities"]

    if expected_count != len(vulnerabilities):
        raise ValueError(
            "CISA feed count does not match the vulnerabilities list length."
        )

    RAW_FILE.write_bytes(raw_bytes)

    with duckdb.connect() as connection:
        top_level_rows = (
            connection.read_json(str(RAW_FILE)).count("*").fetchone()[0]
        )

    print(f"CISA catalog version: {payload['catalogVersion']}")
    print(f"CISA vulnerability records: {expected_count}")
    print(f"DuckDB top-level rows before flattening: {top_level_rows}")
    return expected_count


if __name__ == "__main__":
    extract()

Your raw feed is saved, validated, and visibly nested. Next up, you will flatten its vulnerability list into analytical table rows.

Flatten the Feed into Analytical Tables

Your extractor has already proved that DuckDB sees the complete CISA JSON payload as one top-level row.

That one-row shape keeps every vulnerability inside a nested list. In this step, you will use SQL to flatten that list into one record per vulnerability.

The persistent database will hold raw, staged, and curated tables. Each layer makes the live feed more useful for analysis.

In this step, get ready to:
  • Create a persistent raw table from the downloaded CISA feed.
  • Flatten the nested vulnerability list into typed staged rows.
  • Add analytical flags in a curated table.
Create the persistent raw layer

The raw layer preserves the source payload in its original shape. The database file gives later transformations a stable place to read from.

  • In Visual Studio Code from earlier, select the new-file control in the file sidebar.
  • Name the file transform.py inside cisa-kev-pipeline.
  • Add the imports and path constants to transform.py by pasting this code:
from pathlib import Path

import duckdb

RAW_FILE = Path("data/raw/known_exploited_vulnerabilities.json")
WAREHOUSE_DIR = Path("data/warehouse")
DB_FILE = WAREHOUSE_DIR / "security_pipeline.duckdb"

What Do These Paths Control?

  • The RAW_FILE path points to the live feed saved by extract.py.
  • The WAREHOUSE_DIR path identifies the folder for persistent database storage.
  • The DB_FILE path gives the DuckDB warehouse a consistent filename.
  • Save transform.py.

You should see transform.py listed beside extract.py in the file sidebar.

Cannot See the New File?

Check that you created transform.py inside the cisa-kev-pipeline folder. A file created outside that folder will not share the saved raw data.

Ask for help with finding transform.py in my project folder.

  • Add the raw-layer transformation below the path constants by pasting this code:
def transform():
    if not RAW_FILE.exists():
        raise FileNotFoundError("Run extract.py before transform.py.")

    WAREHOUSE_DIR.mkdir(parents=True, exist_ok=True)

    with duckdb.connect(str(DB_FILE)) as connection:
        connection.execute(
            """
            CREATE OR REPLACE TABLE raw_kev AS
            SELECT *
            FROM read_json('data/raw/known_exploited_vulnerabilities.json');
            """
        )

        raw_count = connection.execute(
            "SELECT count(*) FROM raw_kev"
        ).fetchone()[0]

    print(f"Raw top-level rows: {raw_count}")


if __name__ == "__main__":
    transform()

How Does the Raw Layer Work?

  • The file check stops the transformation if the downloaded CISA feed is missing.
  • The warehouse folder uses repeatable directory creation so another run can reuse it.
  • The database connection creates security_pipeline.duckdb as a persistent local warehouse.
  • The CREATE OR REPLACE TABLE statement rebuilds raw_kev from the current feed.
  • Save transform.py.

Before you run the transformation, do you expect the raw table to contain one row or one row per vulnerability?

  • Create the raw table by running this command in the Visual Studio Code terminal from earlier:
.\.venv\Scripts\python.exe transform.py

What Should You See?

The command runs transform() with the isolated environment from earlier. You should see Raw top-level rows: 1.

That result confirms that raw_kev preserves the original top-level object. The nested vulnerability records still need a staged transformation.

Good progress. Your live source now persists inside security_pipeline.duckdb as the raw layer.

Raw Table Did Not Load?

Confirm that data/raw/known_exploited_vulnerabilities.json still exists inside cisa-kev-pipeline.

Check the indentation beneath with duckdb.connect(str(DB_FILE)) as connection:. The table creation code must remain inside that connection block.

Ask for help with debugging the raw DuckDB table.

Flatten the staged vulnerability rows

The staged layer uses unnest() to turn each item in vulnerabilities into a separate row. Recursive flattening also expands each vulnerability object into columns.

This insertion is a little fiddly because it must stay inside the database connection block. The date casts also convert source strings into DATE values for future comparisons.

  • Inside transform.py, find the line that begins with raw_count = connection.execute(.
  • Insert the staged-table block immediately above that line by pasting this code:
        connection.execute(
            """
            CREATE OR REPLACE TABLE stg_vulnerabilities AS
            SELECT
                cveID AS cve_id,
                vendorProject AS vendor_project,
                product,
                vulnerabilityName AS vulnerability_name,
                CAST(dateAdded AS DATE) AS date_added,
                shortDescription AS short_description,
                requiredAction AS required_action,
                CAST(dueDate AS DATE) AS due_date,
                knownRansomwareCampaignUse AS known_ransomware_campaign_use,
                forensicTriage AS forensic_triage,
                notes,
                cwes
            FROM (
                SELECT unnest(vulnerabilities, recursive := true)
                FROM raw_kev
            );
            """
        )

What Does the Staged Table Change?

The inner query expands the nested list into rows. The outer query gives each source field a readable snake-case name.

The date_added and due_date columns become typed dates. Each CVE now occupies its own analytical row.

  • Find the current count block near the end of transform().
  • Replace that block through the raw-count print statement with this expanded count block:
        raw_count = connection.execute(
            "SELECT count(*) FROM raw_kev"
        ).fetchone()[0]
        staged_count = connection.execute(
            "SELECT count(*) FROM stg_vulnerabilities"
        ).fetchone()[0]

    print(f"Raw top-level rows: {raw_count}")
    print(f"Staged rows: {staged_count}")

Why Count Both Layers?

The raw count confirms that the source remains one top-level object. The staged count reveals how many vulnerability records came out of its nested list.

  • Save transform.py.

Before you run the updated transformation, do you expect the staged count to match the raw count?

  • Build the staged table by running the transformation again:
.\.venv\Scripts\python.exe transform.py

What Should the Counts Show?

You should still see Raw top-level rows: 1. You should also see a Staged rows: count greater than one.

The staged result reflects the current live catalog. Its exact value can change when CISA updates the feed.

That is the nested-data trap solved. The staged count now represents individual vulnerability rows.

Staged Table Failed to Build?

Check that the staged-table block appears before the count queries. Confirm that every line inside the connection block uses the same indentation.

Compare field names such as cveID and vendorProject with the code above. Source field names are case-sensitive.

Ask for help with flattening the vulnerabilities list.

Add curated flags and verify the tables

The curated layer keeps the staged fields while adding decisions that analysts can filter immediately. The remediation_overdue flag compares each due date with the current date.

The ransomware_linked flag identifies rows where the source marker equals Known.

  • Inside transform.py, find the raw_count = connection.execute( line again.
  • Insert the curated-table block immediately above that line by pasting this code:
        connection.execute(
            """
            CREATE OR REPLACE TABLE curated_kev AS
            SELECT
                *,
                due_date < current_date AS remediation_overdue,
                known_ransomware_campaign_use = 'Known' AS ransomware_linked
            FROM stg_vulnerabilities;
            """
        )

What Does the Curated Layer Add?

The SELECT * keeps every staged field. The two boolean expressions add analysis-ready remediation flags.

The table uses CREATE OR REPLACE TABLE so another run rebuilds it from the current staged data.

  • Find the count block that currently reports raw and staged rows.
  • Replace that block with the final three-layer count block:
        raw_count = connection.execute(
            "SELECT count(*) FROM raw_kev"
        ).fetchone()[0]
        staged_count = connection.execute(
            "SELECT count(*) FROM stg_vulnerabilities"
        ).fetchone()[0]
        curated_count = connection.execute(
            "SELECT count(*) FROM curated_kev"
        ).fetchone()[0]

    print(f"Raw top-level rows: {raw_count}")
    print(f"Staged rows: {staged_count}")
    print(f"Curated rows: {curated_count}")
    return curated_count

Why Return the Curated Count?

The three queries expose the row count at every layer. Returning curated_count also lets the complete pipeline reuse the final total later.

  • Save transform.py.

Before you run the completed transformation, which two layers do you expect to have matching counts?

  • Rebuild all three analytical tables by running this command:
.\.venv\Scripts\python.exe transform.py

What Proves the Transformation Worked?

You should see Raw top-level rows: 1. The Staged rows: and Curated rows: values should match.

Both transformed counts should be greater than one. This proves that the nested list became individual analytical records without losing rows between the staged and curated layers.

Counts Do Not Match?

Confirm that curated_kev selects from stg_vulnerabilities without a filter. A missing SELECT * can also leave out staged columns.

Check that each table statement remains inside the same DuckDB connection block. Misplaced indentation can close the connection before later statements run.

Ask for help with comparing my analytical table counts.

You have turned one nested source row into matching staged and curated vulnerability records. Your warehouse now supports typed dates plus remediation-focused filtering.

✔️ Awesome, I've got everything

Great work. Save transform.py once more so your three-layer transformation is ready for the next step.

ⓧ I'd like to double check the full code

Compare your transform.py file with this complete version.

from pathlib import Path

import duckdb

RAW_FILE = Path("data/raw/known_exploited_vulnerabilities.json")
WAREHOUSE_DIR = Path("data/warehouse")
DB_FILE = WAREHOUSE_DIR / "security_pipeline.duckdb"


def transform():
    if not RAW_FILE.exists():
        raise FileNotFoundError("Run extract.py before transform.py.")

    WAREHOUSE_DIR.mkdir(parents=True, exist_ok=True)

    with duckdb.connect(str(DB_FILE)) as connection:
        connection.execute(
            """
            CREATE OR REPLACE TABLE raw_kev AS
            SELECT *
            FROM read_json('data/raw/known_exploited_vulnerabilities.json');
            """
        )

        connection.execute(
            """
            CREATE OR REPLACE TABLE stg_vulnerabilities AS
            SELECT
                cveID AS cve_id,
                vendorProject AS vendor_project,
                product,
                vulnerabilityName AS vulnerability_name,
                CAST(dateAdded AS DATE) AS date_added,
                shortDescription AS short_description,
                requiredAction AS required_action,
                CAST(dueDate AS DATE) AS due_date,
                knownRansomwareCampaignUse AS known_ransomware_campaign_use,
                forensicTriage AS forensic_triage,
                notes,
                cwes
            FROM (
                SELECT unnest(vulnerabilities, recursive := true)
                FROM raw_kev
            );
            """
        )

        connection.execute(
            """
            CREATE OR REPLACE TABLE curated_kev AS
            SELECT
                *,
                due_date < current_date AS remediation_overdue,
                known_ransomware_campaign_use = 'Known' AS ransomware_linked
            FROM stg_vulnerabilities;
            """
        )

        raw_count = connection.execute(
            "SELECT count(*) FROM raw_kev"
        ).fetchone()[0]
        staged_count = connection.execute(
            "SELECT count(*) FROM stg_vulnerabilities"
        ).fetchone()[0]
        curated_count = connection.execute(
            "SELECT count(*) FROM curated_kev"
        ).fetchone()[0]

    print(f"Raw top-level rows: {raw_count}")
    print(f"Staged rows: {staged_count}")
    print(f"Curated rows: {curated_count}")
    return curated_count


if __name__ == "__main__":
    transform()

How the Full File Fits Together

The file preserves the downloaded payload in raw_kev. It flattens that payload into stg_vulnerabilities.

The final transformation adds analytical flags in curated_kev. The printed counts confirm the shape of each layer.

Your three layers now turn the nested CISA feed into typed analytical rows. Next, you will add quality gates before publishing a reusable Parquet data product.

Validate and Export Parquet

Your DuckDB warehouse now turns the nested security feed into typed vulnerability rows. A successful transformation can still lose records or publish incomplete data.

In this step, each risk gets its own data quality gate. Only a fully passing dataset reaches Parquet.

In this step, get ready to:
  • Compare the curated row count with the live source count.
  • Block publication when required values or duplicate CVE IDs fail validation.
  • Publish the validated table as a rerunnable Parquet file.
Create the validation script

The validator begins by connecting to the warehouse from the previous step. Its first measurement compares the source metadata with the curated table.

  • In Visual Studio Code from earlier, create validate_export.py inside the cisa-kev-pipeline folder.
  • Build the validator foundation by pasting this code into validate_export.py:
from pathlib import Path

import duckdb

DB_FILE = Path("data/warehouse/security_pipeline.duckdb")
OUTPUT_DIR = Path("data/curated")
PARQUET_FILE = OUTPUT_DIR / "known_exploited_vulnerabilities.parquet"


def scalar(connection, query):
    return connection.execute(query).fetchone()[0]


def validate_and_export():
    if not DB_FILE.exists():
        raise FileNotFoundError("Run transform.py before validate_export.py.")

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    with duckdb.connect(str(DB_FILE)) as connection:
        source_count = scalar(connection, 'SELECT "count" FROM raw_kev')
        curated_count = scalar(connection, "SELECT count(*) FROM curated_kev")

What does this code do?

  • The path constants identify the persistent warehouse plus the future curated output.
  • The scalar() helper returns one value from a SQL query.
  • The validate_and_export() function confirms that the warehouse exists before reading its counts.
  • Save validate_export.py.
  • Confirm that validate_export.py appears beside your existing Python files in the Visual Studio Code sidebar.

File in the wrong folder?

The file must sit beside extract.py plus transform.py inside cisa-kev-pipeline.

Help me place validate_export.py in the correct project folder..

A row-count gate catches records that disappear between the live feed and the curated table. The script prints the result so every run leaves visible evidence.

  • Complete the first row-count gate by pasting this code directly below the curated_count line:
        checks = {
            "row_count_matches_source": curated_count == source_count,
        }

        for name, passed in checks.items():
            print(f"{name}: {'PASS' if passed else 'FAIL'}")

        failed_checks = [
            name for name, passed in checks.items() if not passed
        ]
        if failed_checks:
            raise ValueError(
                "Data quality checks failed: " + ", ".join(failed_checks)
            )


if __name__ == "__main__":
    validate_and_export()

How does the first gate work?

The checks dictionary records whether both counts match. The loop prints PASS or FAIL for that result.

The failed_checks list stops the function when any gate is false. Invalid data cannot move forward.

  • Save validate_export.py.

Before you run this first gate, what result do you expect? Hold that prediction while you test it.

  • Run the validator from the Visual Studio Code terminal by running this command:
.\.venv\Scripts\python.exe validate_export.py

What should you see?

You should see row_count_matches_source: PASS. This proves the curated table contains the same number of records declared by the live feed.

Does the first gate fail?

A failed row-count gate means the warehouse no longer matches the saved raw feed. Run the extraction and transformation steps again before retrying this validator.

Help me diagnose a failed row-count gate..

Add the remaining quality gates

Matching counts prove that records survived the transformation. Separate checks must still catch missing required values plus repeated vulnerability identifiers.

  • In validate_export.py, select the current checks dictionary.
  • Replace that selection with this expanded quality-gate block:
        required_null_count = scalar(
            connection,
            """
            SELECT count(*)
            FROM curated_kev
            WHERE cve_id IS NULL
               OR vendor_project IS NULL
               OR product IS NULL
               OR due_date IS NULL
            """,
        )
        duplicate_cve_count = scalar(
            connection,
            """
            SELECT count(*) - count(DISTINCT cve_id)
            FROM curated_kev
            """,
        )

        checks = {
            "row_count_matches_source": curated_count == source_count,
            "required_fields_not_null": required_null_count == 0,
            "cve_ids_are_unique": duplicate_cve_count == 0,
        }

What do these gates protect?

  • The required-field query counts rows missing a CVE ID or another field needed for remediation work.
  • The uniqueness query compares all rows with the number of distinct CVE IDs.
  • The expanded checks dictionary requires every invalid count to equal zero.
  • Save validate_export.py.

Before you rerun the validator, do you expect all three gates to pass? Keep your prediction in mind while the queries run.

  • Test all three quality gates by running this command:
.\.venv\Scripts\python.exe validate_export.py

What should you see now?

You should see PASS for row_count_matches_source, required_fields_not_null, plus cve_ids_are_unique.

That is the quality barrier working. The dataset now proves its record count plus the integrity of its required identifiers.

Seeing a failed quality gate?

Use the failed gate name to narrow the problem. A required-field failure points to null values in the curated columns.

A uniqueness failure points to repeated cve_id values. Rebuild the warehouse from the current raw feed before testing again.

Help me investigate the failed quality gate..

Export and verify the Parquet file

The pipeline can publish its data product once every database check passes. Reading the new file back verifies that the export contains the same number of rows.

  • In validate_export.py, find the if __name__ == "__main__": guard.
  • Place your cursor on the blank line immediately above that guard.
  • Add the Parquet export plus its verification by pasting this code:
        connection.table("curated_kev").write_parquet(
            str(PARQUET_FILE), overwrite=True
        )
        parquet_count = (
            connection.read_parquet(str(PARQUET_FILE))
            .count("*")
            .fetchone()[0]
        )

        parquet_check = parquet_count == curated_count
        print(
            "parquet_count_matches_curated: "
            + ("PASS" if parquet_check else "FAIL")
        )
        if not parquet_check:
            raise ValueError("Parquet row count does not match curated_kev.")

    print(f"Parquet rows: {parquet_count}")
    print(f"Parquet output: {PARQUET_FILE}")
    return parquet_count

How is the export verified?

  • The write_parquet() call publishes curated_kev to the curated data folder.
  • The overwrite=True setting lets future runs replace the existing file.
  • The read-back check counts the rows inside the published file.
  • The final error prevents a mismatched file from being treated as a valid data product.
  • Save validate_export.py.

✔️ Awesome, I've got everything!

Your validator now checks the warehouse before publishing its verified Parquet output.

ⓧ I'd like to double check the full code

from pathlib import Path

import duckdb

DB_FILE = Path("data/warehouse/security_pipeline.duckdb")
OUTPUT_DIR = Path("data/curated")
PARQUET_FILE = OUTPUT_DIR / "known_exploited_vulnerabilities.parquet"


def scalar(connection, query):
    return connection.execute(query).fetchone()[0]


def validate_and_export():
    if not DB_FILE.exists():
        raise FileNotFoundError("Run transform.py before validate_export.py.")

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    with duckdb.connect(str(DB_FILE)) as connection:
        source_count = scalar(connection, 'SELECT "count" FROM raw_kev')
        curated_count = scalar(connection, "SELECT count(*) FROM curated_kev")
        required_null_count = scalar(
            connection,
            """
            SELECT count(*)
            FROM curated_kev
            WHERE cve_id IS NULL
               OR vendor_project IS NULL
               OR product IS NULL
               OR due_date IS NULL
            """,
        )
        duplicate_cve_count = scalar(
            connection,
            """
            SELECT count(*) - count(DISTINCT cve_id)
            FROM curated_kev
            """,
        )

        checks = {
            "row_count_matches_source": curated_count == source_count,
            "required_fields_not_null": required_null_count == 0,
            "cve_ids_are_unique": duplicate_cve_count == 0,
        }

        for name, passed in checks.items():
            print(f"{name}: {'PASS' if passed else 'FAIL'}")

        failed_checks = [
            name for name, passed in checks.items() if not passed
        ]
        if failed_checks:
            raise ValueError(
                "Data quality checks failed: " + ", ".join(failed_checks)
            )

        connection.table("curated_kev").write_parquet(
            str(PARQUET_FILE), overwrite=True
        )
        parquet_count = (
            connection.read_parquet(str(PARQUET_FILE))
            .count("*")
            .fetchone()[0]
        )

        parquet_check = parquet_count == curated_count
        print(
            "parquet_count_matches_curated: "
            + ("PASS" if parquet_check else "FAIL")
        )
        if not parquet_check:
            raise ValueError("Parquet row count does not match curated_kev.")

    print(f"Parquet rows: {parquet_count}")
    print(f"Parquet output: {PARQUET_FILE}")
    return parquet_count


if __name__ == "__main__":
    validate_and_export()

How to use this reference

This is the complete validate_export.py file for this step. Compare its indentation plus line order with your saved file.

Before the final run, how many passing checks do you expect to see? Keep that number in mind while the validator publishes the file.

  • Run the completed validator from the cisa-kev-pipeline folder by running this command:
.\.venv\Scripts\python.exe validate_export.py

What proves the export worked?

You should see PASS for the row-count check. You should also see PASS for the required-field check.

The CVE-uniqueness check plus parquet_count_matches_curated should also print PASS.

The final lines show the current row count plus data\curated\known_exploited_vulnerabilities.parquet. Your validated data product is ready for downstream analysis.

Parquet check not passing?

Confirm that the file path uses data/curated/known_exploited_vulnerabilities.parquet. A different location can make the read-back check inspect the wrong file.

Confirm that overwrite=True remains in the export call. This lets the validator replace an older output.

Help me diagnose the Parquet export check..

Your pipeline now refuses to publish unverified data. Next up, you will connect every stage into one command and prove that the complete batch job runs cleanly twice.

Run the Complete Pipeline Twice

Your extractor now downloads the live catalog. Your DuckDB tables now publish a validated Parquet file.

A batch job earns trust when one command can rebuild every output without manual cleanup. The second run proves that the existing outputs can be replaced cleanly.

In this step, get ready to:
  • Create pipeline.py to coordinate the completed stages.
  • Run the full pipeline from the Visual Studio Code terminal.
  • Repeat the command to prove the outputs rebuild cleanly.
Create the pipeline runner

The runner gives your batch ETL process one entry point. It reuses the functions you already tested.

  • Use New File in the Explorer sidebar to add the code below as pipeline.py inside cisa-kev-pipeline:
from extract import extract
from transform import transform
from validate_export import validate_and_export


def run_pipeline():
    print("Starting CISA KEV batch pipeline")
    extract()
    transform()
    exported_rows = validate_and_export()
    print(f"Pipeline complete: {exported_rows} validated rows exported")


if __name__ == "__main__":
    run_pipeline()

What does this code do?

  • The imports make each completed stage available to pipeline.py.
  • The run_pipeline() function starts with a fresh extraction.
  • The transform() call recreates each warehouse table from the downloaded feed.
  • The validate_and_export() call checks the rebuilt data before publishing the Parquet output.
  • The exported_rows value records how many validated rows reached the output file.
  • Save pipeline.py.
Run the first complete rebuild

The first end-to-end run establishes your baseline. One process now refreshes the source before rebuilding every downstream output.

Before you run the pipeline, which completed stage do you expect to print its results first?

  • Run the complete pipeline from the terminal by using this command:
.\.venv\Scripts\python.exe pipeline.py

You'll see Raw top-level rows: 1 after the live catalog details. All four checks print PASS before the final line containing Pipeline complete.

Pipeline Run Stopped?

  • Confirm your Windows computer can reach the public internet if the download stops.
  • Confirm that extract.py sits beside pipeline.py if the extraction import fails.
  • Confirm that transform.py sits beside validate_export.py if a later import fails.

Help me diagnose why my CISA KEV pipeline stopped during its first complete run.

You now have the full batch flow working from a single command.

✔️ Awesome, I've got everything!

Your saved pipeline.py matches the orchestration flow you just ran.

ⓧ I'd like to double check the full code

from extract import extract
from transform import transform
from validate_export import validate_and_export


def run_pipeline():
    print("Starting CISA KEV batch pipeline")
    extract()
    transform()
    exported_rows = validate_and_export()
    print(f"Pipeline complete: {exported_rows} validated rows exported")


if __name__ == "__main__":
    run_pipeline()
Repeat the pipeline without cleanup

A rerunnable pipeline replaces yesterday's outputs with a fresh build. Your existing database tables and Parquet file now provide the test.

Before you run it again, do you expect an existing table or output file to stop the pipeline?

  • Run the same complete pipeline a second time by using this command:
.\.venv\Scripts\python.exe pipeline.py

You'll see all four checks print PASS again. The second run ends with Pipeline complete without a cleanup error.

  • Compare the CISA vulnerability count with the staged count from each run.
  • Compare the curated count with the Parquet count from each run.

The matching counts prove that each run rebuilt the current feed cleanly. The values across runs can differ if CISA publishes an update between requests.

Second Run Failing?

  • Close any application that has security_pipeline.duckdb open if the second run reports a file-access problem.
  • Confirm that transform.py still uses CREATE OR REPLACE TABLE for every warehouse layer.
  • Confirm that validate_export.py still passes overwrite=True when writing the Parquet file.

Help me find why my pipeline succeeds once but fails on its second run.

That's the repeatability test passed. Your pipeline now refreshes the feed before recreating its validated warehouse tables and Parquet data product.

Secret mission

Export the Ransomware-Linked Queue

Create a focused Parquet queue for vulnerabilities linked to known ransomware campaigns. Keep the pipeline rerunnable while sorting the queue by due date and CVE ID.

Clean Up Your Resources

Clean Up Your Resources

Every resource lives inside your local cisa-kev-pipeline folder, where it uses disk space without creating cloud charges. Choose whether to keep everything, pause by removing .venv, or delete the folder.

Resources you used:

  • The cisa-kev-pipeline folder with your Python scripts and pinned requirements.txt file.
  • The downloaded data/raw/known_exploited_vulnerabilities.json source from the CISA Known Exploited Vulnerabilities Catalog.
  • The data/warehouse/security_pipeline.duckdb warehouse with the DuckDB tables raw_kev, stg_vulnerabilities, and curated_kev.
  • The validated Parquet export at data/curated/known_exploited_vulnerabilities.parquet.
  • The optional Secret Mission export at data/curated/ransomware_linked.parquet.
  • The .venv environment containing DuckDB 1.5.5.

Keep everything running

Your local pipeline stays ready to rerun without creating cloud costs.

  • Retain the cisa-kev-pipeline folder in its current location.
  • Keep the raw feed, warehouse, and curated Parquet exports for future analysis.
  • Leave .venv in place so DuckDB remains available.
  • Use the pipeline again whenever you need a refreshed CISA catalog.

Pause - I'll come back to this later

Pausing removes the replaceable environment while preserving your code and generated data.

  • In the Visual Studio Code Explorer sidebar, right-click .venv.
  • Choose the delete option.
  • Confirm that .venv disappears while your scripts and data folder remain.
  • Rebuild the isolated environment when you return by running these commands in the Visual Studio Code terminal:
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install -r requirements.txt

What do these commands restore?

The first command creates a fresh isolated environment.

The second reads requirements.txt. It installs DuckDB 1.5.5 into the new environment.

  • Confirm the restored DuckDB version by running this command:
.\.venv\Scripts\python.exe -c "import duckdb; print(duckdb.__version__)"

What should I see?

The terminal prints 1.5.5.

That confirms your isolated DuckDB environment is ready for another pipeline run.

Environment not rebuilding?

  • Check that the terminal path ends with cisa-kev-pipeline.
  • Confirm that requirements.txt contains duckdb==1.5.5.
  • Get help diagnosing why the isolated environment cannot be rebuilt.

Delete - I don't want to use this again

Deleting the folder can feel final. Copy anything you want to keep before removing every resource listed above.

  • In the Visual Studio Code Explorer sidebar, right-click cisa-kev-pipeline.
  • Choose the option that reveals the folder in File Explorer.
  • Close the Visual Studio Code window from earlier.
  • In File Explorer, right-click cisa-kev-pipeline.
  • Choose the delete option.

The folder moves to the Recycle Bin with its raw feed, DuckDB warehouse, Parquet exports, scripts, and .venv environment.

  • Open the Recycle Bin.
  • Right-click cisa-kev-pipeline.
  • Choose the permanent delete option.
  • Confirm the deletion when Windows asks.

You should no longer see cisa-kev-pipeline in the Recycle Bin. That confirms every local project resource is gone.

Nice Work!

Nice Work!

You did it! Your local Batch ETL pipeline now turns CISA's live Known Exploited Vulnerabilities Catalog into validated DuckDB tables plus reusable Parquet files.

You've learned how to:

  • Used HTTP extraction to download the live catalog. Exposed the nested JSON trap by comparing the feed's vulnerability count with DuckDB's single top-level row.
  • Built a layered warehouse in DuckDB. raw_kev preserves the source. stg_vulnerabilities flattens each vulnerability. curated_kev adds typed dates plus remediation flags.
  • Added quality gates for source row counts. A required-field check catches missing values. A uniqueness check catches duplicate CVE IDs. Published a rerunnable Parquet data product only after every check passed.
  • Secret Mission: Published a due-date-sorted ransomware-linked remediation queue to data/curated/ransomware_linked.parquet.

Ready to quiz yourself?