Security Log ETL Pipeline with dbt

Build a threat-detection pipeline using dbt, PostgreSQL, and Streamlit.

Introduction

30 Second Summary

Every organization generates thousands of security events daily. When this data is messy and scattered, real threats hide in plain sight.

In this project, you will build a complete ELT pipeline for security threat detection. You will generate raw log data, load it into PostgreSQL, transform it with dbt, and visualize the results in a live Streamlit dashboard.

What You'll Build

You will run a live dashboard that surfaces brute-force login patterns and after-hours access anomalies from 5,000 security events, all processed through a pipeline you built from scratch.

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

  • A Python data generator that creates 5,000 realistic security events with intentional quality issues and loads them into PostgreSQL via SQLAlchemy, so you can see firsthand how messy data breaks analytics.
  • A dbt project with staging and mart models that clean raw data and detect brute-force login attempts and suspicious after-hours access using window functions and CTEs.
  • An interactive Streamlit dashboard with KPI metrics, threat data tables, and bar charts showing event distribution and risk levels pulled live from your analytics schema.
  • Secret Mission: Add interactive date-range filtering and event-type drill-down to turn your dashboard into an investigation tool.

Are there any prerequisites?

This project assumes you have Python 3.10+ and PostgreSQL installed locally. You should be comfortable writing SQL queries, including joins and aggregations.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Project Environment

Your ELT pipeline needs three foundations: a Python environment for the Extract and Load scripts, a PostgreSQL database to hold the data, and a dbt project to manage the Transform layer.

In this step, you'll set all three up. By the end, dbt will confirm it can talk to your database, proving every piece is connected and ready.

In this step, get ready to:
  • Create a Python virtual environment and install all project dependencies.
  • Set up a PostgreSQL database with a dedicated analytics schema.
  • Initialize a dbt project and verify the database connection.
Create your project directory and virtual environment
  • Move to your Desktop and create the project folder by running these commands:
cd ~/Desktop
mkdir security-etl
cd security-etl
  • Create a Python virtual environment by running:
python -m venv .venv
  • Activate the virtual environment:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

Your terminal prompt now shows (.venv) at the beginning. This confirms the virtual environment is active and all packages you install will stay isolated to this project.

Don't see (.venv) in your prompt?

Make sure you ran the correct activation command for your operating system. On macOS, the path uses bin/activate. On Windows, the path uses Scripts\activate.

If you see a permissions error on Windows, try running your terminal as Administrator.

help me activate my Python virtual environment

  • Install all project dependencies by running these two commands:
pip install pandas==2.2.3 psycopg2-binary==2.9.12 SQLAlchemy==2.0.52 streamlit==1.62.0
pip install "dbt-core<2.0" dbt-postgres

Why two separate install commands?

The first command installs your core Python libraries with pinned versions: pandas for data manipulation, psycopg2-binary for PostgreSQL connectivity, SQLAlchemy for database engine management, and Streamlit for the dashboard.

The second command installs dbt with a version constraint. The "dbt-core<2.0" pin prevents pip from pulling dbt-core 2.0 alpha, which breaks the PostgreSQL adapter.

  • Verify all libraries installed correctly by running:
python -c "import pandas; import sqlalchemy; import psycopg2; import streamlit; print('All imports OK')"

You should see All imports OK printed to the terminal.

  • Confirm dbt installed correctly by running:
dbt --version

You should see a 1.x version number in the output.

Seeing an import error or command not found?

Check that (.venv) still appears in your terminal prompt. If it disappeared, reactivate the virtual environment using the activation command for your operating system.

If a specific package failed to install, re-run the pip install command for that package on its own to see the full error message.

help me fix my Python package installation

Create the PostgreSQL database and analytics schema

Your pipeline needs a dedicated database to keep security log data separate from any other PostgreSQL data on your machine. You'll also create an analytics schema where dbt will write its transformed models.

  • Open a psql session by running:
psql -U postgres

Adjust for your PostgreSQL setup

If your PostgreSQL user is not postgres, replace it with your own username. On macOS, you may only need psql postgres if you installed via Homebrew.

  • Create the database and schema by running these commands inside psql:
CREATE DATABASE security_logs;
\c security_logs
CREATE SCHEMA analytics;

What do these commands do?

CREATE DATABASE security_logs; creates a brand new database dedicated to this project.

\c security_logs switches your psql session into that database so the next command runs inside it.

CREATE SCHEMA analytics; creates a named schema where dbt will write all its transformed views and tables.

  • Verify the schema was created by running:
\dn

You should see analytics listed alongside the default public schema.

  • Exit psql by running:
\q

Connection refused or role does not exist?

Make sure your PostgreSQL server is running. On macOS, check with brew services list. On Windows, search for Services in the Start menu and confirm postgresql is running.

If you see a 'role does not exist' error, try connecting with your operating system username instead of postgres.

help me connect to my local PostgreSQL database

Initialize the dbt project and verify the connection

With the database ready, you need a dbt project to define your SQL transformations. The dbt init command scaffolds a project folder with the standard directory structure dbt expects.

  • Make sure you are back in the security-etl directory with your virtual environment active. Initialize the dbt project by running:
dbt init security_analytics -s

What happens when you run this?

dbt will ask you to choose a database adapter. Type the number next to postgres and press Enter. The -s flag skips the interactive profile setup so you can create the connection profile manually with exact control over every value.

After selecting the adapter, dbt creates a security_analytics/ folder with sample models, a default dbt_project.yml, and the directory structure your project needs.

  • Confirm the project was created by running:
ls security_analytics

You should see folders like models, macros, tests, and a dbt_project.yml file.

Now you need to tell dbt how to connect to your security_logs database. dbt reads connection details from a file called profiles.yml stored in a .dbt directory in your home folder.

  • Create the .dbt directory if it does not already exist:

macOS

mkdir -p ~/.dbt

Windows

mkdir "$env:USERPROFILE\.dbt" -Force
  • Create a file called profiles.yml inside the .dbt directory and paste in these connection details:
security_analytics:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: your-username
      password: your-password
      dbname: security_logs
      schema: analytics
      threads: 4
  • Replace your-username with the PostgreSQL username you used to connect via psql earlier.
  • Replace your-password with that user's password. If your local PostgreSQL has no password, leave the value as an empty string.

What does profiles.yml configure?

This file maps the profile name security_analytics to a specific database connection. The schema: analytics line tells dbt to create all models inside the analytics schema you just created.

The threads: 4 setting lets dbt run up to four models in parallel, which speeds up builds as the project grows.

  • Next, configure the dbt project file. Open security_analytics/dbt_project.yml and replace its entire contents with:
name: 'security_analytics'
version: '1.0.0'

profile: 'security_analytics'

model-paths: ["models"]
seed-paths: ["seeds"]
test-paths: ["tests"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

clean-targets:
  - "target"
  - "dbt_packages"

models:
  security_analytics:
    staging:
      +materialized: view
    marts:
      +materialized: table

What does the models section do?

The models block at the bottom pre-configures how dbt materializes your SQL. Staging models become views (lightweight, always fresh), while mart models become tables (faster to query, great for dashboards).

You have not created these model files yet. This configuration will apply automatically once you do.

Before you run the connection test, what do you think dbt debug will check? Consider what could go wrong between dbt and your database.

  • Move into the dbt project directory and test the connection by running:
cd security_analytics
dbt debug

You should see a series of checks ending with All checks passed! confirming dbt can find your profile, connect to PostgreSQL, and reach the security_logs database.

dbt debug showing a connection failure?

Double check that the user and password values in profiles.yml match exactly what you used with psql. Also confirm dbname is set to security_logs and schema is set to analytics.

Make sure your PostgreSQL server is still running. The database you created earlier must be accessible on localhost:5432.

help me fix my dbt debug connection failure

✔️ Awesome, I've got everything!

Great. Double check you saved both profiles.yml and dbt_project.yml.

ⓧ I'd like to double check the full code

Here are both configuration files. Make sure yours match exactly.

~/.dbt/profiles.yml:

security_analytics:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: your-username
      password: your-password
      dbname: security_logs
      schema: analytics
      threads: 4

security_analytics/dbt_project.yml:

name: 'security_analytics'
version: '1.0.0'

profile: 'security_analytics'

model-paths: ["models"]
seed-paths: ["seeds"]
test-paths: ["tests"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

clean-targets:
  - "target"
  - "dbt_packages"

models:
  security_analytics:
    staging:
      +materialized: view
    marts:
      +materialized: table

Your environment is locked in. Python, PostgreSQL, and dbt are all connected and ready. Next up, you'll generate realistic security log data and load it into your database.

Generate and Load Raw Security Logs

Your project environment is set up and dbt can talk to PostgreSQL. Before you can build any transforms or analytics, you need data in the database to work with.

In this step, you'll write a Python script that generates 5,000 synthetic security events with intentional data quality issues, then loads them into PostgreSQL using SQLAlchemy. This covers the Extract and Load half of your ELT pipeline.

In this step, get ready to:
  • Create a Python script that generates realistic security event data with quality issues.
  • Load 5,000 records into PostgreSQL using pandas and SQLAlchemy.
  • Query the raw data in psql to see the messy records firsthand.
Create the data generator script

Security logs in the real world are messy. Usernames arrive in mixed case, event types aren't standardized, and some fields are missing entirely. Your generator will replicate these problems on purpose so you have realistic data to clean later with dbt.

  • Create a new file called generate_logs.py in the security-etl folder (not inside security_analytics).
  • Add the imports, database connection string, and the first two data pools by pasting this code:
import pandas as pd
import random
from datetime import datetime, timedelta
from sqlalchemy import create_engine

# Connection string for your local PostgreSQL database
DB_URL = "postgresql+psycopg2://your-username:your-password@localhost:5432/security_logs"
NUM_RECORDS = 5000

random.seed(42)

# Usernames with intentional quality issues: mixed casing, empty strings, None
usernames = [
    "jsmith", "JSMITH", "j.smith", "admin", "Admin", "ADMIN",
    "mgarcia", "MGarcia", "lchen", "lChen", "root", "svc_backup",
    "deploy_bot", "null_user", "", None,
]

# Event types with inconsistent casing
event_types = [
    "LOGIN_SUCCESS", "login_success", "LOGIN_FAILURE", "login_failure",
    "PRIVILEGE_ESCALATION", "privilege_escalation",
    "FILE_ACCESS", "file_access", "LOGOUT", "logout",
    "PASSWORD_CHANGE", "ACCOUNT_LOCKED",
]

What does this code do?

DB_URL tells SQLAlchemy where to find your PostgreSQL database. Replace your-username and your-password with the credentials you use to connect to PostgreSQL locally.

The usernames and event_types lists are data pools. Each entry is a possible value the generator can pick from randomly. Notice they include the same person in different casing (jsmith vs JSMITH vs j.smith), empty strings, and None values. These are intentional quality issues you'll clean up later.

  • Add the remaining three data pools below the event_types list:
# Source IPs including invalid entries
source_ips = [
    "192.168.1.100", "192.168.1.101", "10.0.0.50", "10.0.0.51",
    "172.16.0.10", "203.0.113.42", "198.51.100.7", "192.0.2.1",
    "UNKNOWN", "", None, "not_an_ip",
]

# Departments with mixed casing, None, and empty strings
departments = [
    "Engineering", "engineering", "ENGINEERING", "Finance", "finance",
    "HR", "hr", "Security", "IT", None, "",
]

# Risk levels with inconsistent casing and None
risk_levels = [
    "low", "LOW", "Low", "medium", "MEDIUM", "Medium",
    "high", "HIGH", "High", "critical", "CRITICAL", None,
]

What's happening here?

Each pool mirrors a real-world data problem. source_ips includes values like UNKNOWN and not_an_ip that aren't valid IP addresses. departments and risk_levels have the same values in different cases (Engineering vs ENGINEERING). These are the kinds of inconsistencies that break analytics queries.

  • Save generate_logs.py.
  • Confirm the file exists by running:
ls generate_logs.py

You should see generate_logs.py in the output.

Don't see the file?

Make sure you created the file inside the security-etl folder, not inside security_analytics. Run pwd to check your current directory.

help me find where I created generate_logs.py

  • Now add the two generator functions below the risk_levels list:
# Generate a random timestamp within the first half of 2024
def generate_timestamp():
    base = datetime(2024, 1, 1)
    offset = timedelta(
        days=random.randint(0, 180),
        hours=random.randint(0, 23),
        minutes=random.randint(0, 59),
        seconds=random.randint(0, 59),
    )
    return base + offset


# Build n event records by randomly sampling from each data pool
def generate_records(n):
    records = []
    for _ in range(n):
        records.append(
            {
                "event_timestamp": generate_timestamp(),
                "username": random.choice(usernames),
                "event_type": random.choice(event_types),
                "source_ip": random.choice(source_ips),
                "department": random.choice(departments),
                "risk_level": random.choice(risk_levels),
                "success": random.choice([True, False, None]),
                "session_duration_sec": random.choice(
                    [random.randint(1, 28800), -1, 0, None]
                ),
            }
        )
    return records

What does this code do?

generate_timestamp() creates a random date and time between January 1 and June 29, 2024. This gives you a realistic spread of security events over six months.

generate_records() builds a list of dictionaries. Each dictionary is one security event. Notice session_duration_sec can be -1, 0, or None. Negative durations are another intentional quality issue.

  • Finally, add the main() function and entry point below generate_records:
# Generate the data, preview it, and load into PostgreSQL
def main():
    print("Generating security log data...")
    records = generate_records(NUM_RECORDS)
    df = pd.DataFrame(records)

    print(f"Generated {len(df)} records")
    print(f"\nSample data:")
    print(df.head(10).to_string())
    print(f"\nData types:\n{df.dtypes}")
    print(f"\nNull counts:\n{df.isnull().sum()}")

    # Connect to PostgreSQL and write the DataFrame as a table
    engine = create_engine(DB_URL)
    df.to_sql("raw_security_events", engine, if_exists="replace", index=False)
    print(f"\nLoaded {len(df)} records into 'raw_security_events' table.")


if __name__ == "__main__":
    main()

What does this code do?

main() generates all 5,000 records, wraps them in a pandas DataFrame, and prints a preview of the data, its types, and null counts to your terminal.

df.to_sql() writes the entire DataFrame into a PostgreSQL table called raw_security_events. The if_exists="replace" parameter drops and recreates the table each time you run the script, so you can safely rerun it without duplicating data.

  • Save generate_logs.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved generate_logs.py and that you replaced your-username and your-password in the DB_URL string with your actual PostgreSQL credentials.

ⓧ I'd like to double check the full code

Scroll up and confirm each of these sections is present in your file, in order:

  • Imports, DB_URL, NUM_RECORDS, and random.seed(42).
  • usernames list (16 entries including empty string and None).
  • event_types list (12 entries with mixed casing).
  • source_ips list (12 entries including UNKNOWN and not_an_ip).
  • departments list (11 entries).
  • risk_levels list (12 entries).
  • generate_timestamp() function.
  • generate_records(n) function.
  • main() function with to_sql() call.
  • if __name__ == "__main__": block at the very bottom.
Run the generator and explore the raw data

The script is ready. Before you run it, think about what you expect to see in the terminal output given the data pools you defined.

  • Run the data generator by running this command:
python generate_logs.py

You should see a sample of 10 records, the data types for each column, null counts per column, and a final line confirming 5,000 records were loaded.

What should I see?

The sample data shows usernames like JSMITH next to j.smith, event types in mixed casing, and source IPs that include not_an_ip. The null counts show that several columns have missing values. This is exactly the messy data you'll clean with dbt in the next step.

Seeing a connection error?

Make sure your PostgreSQL server is running. Check that the DB_URL in generate_logs.py uses the same username, password, and port that you used in your profiles.yml from the previous step.

help me debug my PostgreSQL connection in generate_logs.py

Now let's look at the data directly in the database. Connect to your security_logs database in psql.

  • Open a psql session by running:
psql -d security_logs
  • Query a sample of the raw data by running:
SELECT username, event_type, source_ip FROM raw_security_events LIMIT 10;

You should see rows with mixed-case usernames, inconsistent event type casing, and source IPs that include invalid entries like empty strings or not_an_ip.

Before you run the next query, how many total rows do you think the table has?

  • Verify the row count by running:
SELECT count(*) FROM raw_security_events;

The result is 5000. All 5,000 synthetic security events are loaded and ready for transformation.

  • Exit psql by typing \q and pressing Enter.

Your raw security logs are loaded into PostgreSQL. Next, you'll discover just how badly these quality issues break your analytics when you try to query the data for threat detection.

Discover Dirty Data and Build dbt Staging Models

Your raw security events are sitting in PostgreSQL, but there is a hidden problem. The same user appears under multiple name variants like jsmith, JSMITH, and j.smith. Any threat detection query you run will treat them as separate people and produce wrong results.

In this step, you will see this problem firsthand by querying the raw data. Then you will build a dbt staging model that cleans and normalizes everything using SQL transforms that run inside PostgreSQL.

In this step, get ready to:
  • Query raw data to discover how inconsistent formatting breaks analytics.
  • Build a dbt staging model that cleans and normalizes the raw data.
  • Run dbt tests to validate data quality automatically.
Query the raw data to see the problem

Before building any transforms, you need to see exactly how the messy data breaks a real analytics query. A simple GROUP BY on usernames will reveal the problem.

Before you run this query, think about what you expect to see. If a user named "jsmith" logged in 50 times, would you expect one row with a count of 50, or something else?

  • Switch back to your psql session connected to security_logs and run this query:
SELECT username, count(*) FROM raw_security_events GROUP BY username ORDER BY count(*) DESC LIMIT 10;

You should see the same person appearing as multiple separate rows. jsmith, JSMITH, and j.smith each have their own count instead of being combined into one.

Why does this break threat detection?

If you are trying to detect brute-force attacks (5+ failed logins from the same user in an hour), splitting "jsmith" across three rows means the failures for that user are divided across three separate counts. An attacker who triggers 15 failures would show up as three entries of 5 instead of one entry of 15.

This is the exact problem that dbt staging models solve. You will write SQL that normalizes the data inside PostgreSQL before any analytics queries touch it.

Set up the staging model directory structure

The dbt project created an example models folder during initialization. You need to replace it with staging and marts directories that match the ELT pattern.

  • In your terminal, make sure you are inside the security_analytics directory. Delete the example folder and create the new directories by running these commands:
cd security_analytics
rm -rf models/example
mkdir models/staging models/marts
  • Confirm the directories exist by running:
ls models/

You should see staging and marts listed. The example folder should be gone.

Still seeing the example folder?

Make sure you ran the commands from inside the security_analytics directory, not the parent security-etl folder. Run pwd to check your current location.

Need more help? Help me find the correct directory for my dbt project

Create the staging model source definition and SQL

A dbt source tells dbt where your raw data lives. You need a source definition file that points to the raw_security_events table in the public schema.

  • Create a new file called _sources.yml inside models/staging/ and paste the following:
version: 2

sources:
  - name: raw
    schema: public
    tables:
      - name: raw_security_events
        description: "Raw security event logs loaded by the Python generator"

What does this file do?

This YAML file registers your raw_security_events table as a dbt source named raw. Your SQL models can now reference this table using {{ source('raw', 'raw_security_events') }} instead of hardcoding the table name.

Now you will create the staging model SQL file. This model applies cleaning and normalization transforms to every column. Because it is a long query, you will build it in two parts.

  • Create a new file called stg_security_events.sql inside models/staging/ and paste the first part of the query:
with source as (

    select * from {{ source('raw', 'raw_security_events') }}

),

cleaned as (

    select
        event_timestamp,
        lower(trim(coalesce(nullif(username, ''), 'unknown'))) as username,
        upper(trim(coalesce(nullif(event_type, ''), 'UNKNOWN'))) as event_type,
        case
            when source_ip ~ '^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$'
                then source_ip
            else null
        end as source_ip,

What does this code do?

  • The source CTE pulls all rows from raw_security_events using the dbt source reference you defined in _sources.yml.
  • lower(trim(coalesce(nullif(username, ''), 'unknown'))) works from the inside out: nullif converts empty strings to NULL, coalesce replaces NULLs with 'unknown', trim removes whitespace, and lower lowercases everything.
  • The regex check (~) validates that source_ip matches the IPv4 pattern. Invalid values like UNKNOWN and not_an_ip become NULL instead of polluting your results.
  • Confirm both files exist in models/staging/ by running:
ls models/staging/

You should see _sources.yml and stg_security_events.sql listed.

  • Now add the remaining columns to stg_security_events.sql. Paste this code directly below the end as source_ip, line:
        lower(trim(coalesce(nullif(department, ''), 'unknown'))) as department,
        lower(trim(coalesce(nullif(risk_level, ''), 'unknown'))) as risk_level,
        success,
        case
            when session_duration_sec is not null
                 and session_duration_sec > 0
                then session_duration_sec
            else null
        end as session_duration_sec,
        extract(hour from event_timestamp)::int as event_hour,
        extract(dow from event_timestamp)::int as day_of_week
    from source

)

select * from cleaned

What does this code do?

  • department and risk_level get the same lower/trim/coalesce/nullif treatment as username.
  • session_duration_sec drops negative and zero values (the raw data has -1 and 0 entries that represent bad data).
  • event_hour and day_of_week are extracted from the timestamp for time-based analysis. extract(dow ...) returns 0 for Sunday through 6 for Saturday in PostgreSQL.
  • Save stg_security_events.sql.
  • Build the staging model by running:
dbt run

You should see output showing that the stg_security_events model was created successfully as a view in the analytics schema.

Seeing a compilation error?

  • Check that the two code chunks join seamlessly. The line ending in end as source_ip, should be immediately followed by the lower(trim(...department...)) line with no extra commas or missing commas between them.
  • Make sure the file ends with select * from cleaned and the closing parenthesis ) appears on its own line before that final select.
  • Verify that _sources.yml is saved in models/staging/ and not in the project root.

Help me debug my dbt run error for the staging model

✔️ Awesome, I've got everything!

Great. Double check you have saved stg_security_events.sql and _sources.yml.

ⓧ I'd like to double check the full code

Your _sources.yml should look exactly like this:

version: 2

sources:
  - name: raw
    schema: public
    tables:
      - name: raw_security_events
        description: "Raw security event logs loaded by the Python generator"

Your stg_security_events.sql combines both chunks. Scroll up and confirm you have the source CTE pulling from the raw source, then the cleaned CTE with all columns (event_timestamp, username, event_type, source_ip, department, risk_level, success, session_duration_sec, event_hour, day_of_week), and a final select * from cleaned at the bottom.

Add dbt tests and verify the clean data

dbt has a built-in testing framework that validates your data after every run. You will add not_null tests to catch any rows where your cleaning logic missed a null value.

  • Create a new file called schema.yml inside models/staging/ and paste the following:
version: 2

models:
  - name: stg_security_events
    description: "Cleaned and normalized security events"
    columns:
      - name: username
        description: "Lowercase, trimmed username"
        tests:
          - not_null
      - name: event_type
        description: "Uppercase, standardized event type"
        tests:
          - not_null
      - name: event_timestamp
        description: "Timestamp of the event"
        tests:
          - not_null

What does this file do?

This schema file defines three not_null tests on username, event_type, and event_timestamp. Every time you run dbt test, dbt will check that none of these columns contain NULL values in the cleaned staging view.

  • Save schema.yml and run the tests:
dbt test

All three tests should pass with green checkmarks.

Seeing test failures?

  • A failing not_null test means your SQL transform is not catching all NULL or empty string values. Double check the coalesce(nullif(...)) wrapping on the failing column in stg_security_events.sql.
  • Make sure the column names in schema.yml match the aliases in your SQL exactly (e.g. username, not user_name).

Help me debug my failing dbt tests

Now verify that the cleaned data actually fixes the username problem you saw earlier. Do you think the GROUP BY query will still show duplicate usernames?

  • Switch back to your psql session and run this query against the staging view:
SELECT DISTINCT username FROM analytics.stg_security_events ORDER BY username LIMIT 10;

All usernames should now be lowercase and consistent. No more jsmith vs JSMITH vs j.smith appearing as separate users.

Your raw data is now cleaned and normalized inside PostgreSQL. Next up, you will build mart models that use this clean data to detect actual threats like brute-force login attempts and suspicious after-hours access.

Build Threat Detection Mart Models

Your staging model is doing the heavy lifting on data quality. Every username is lowercase, every event type is uppercase, and invalid IPs are gone. But clean data alone does not catch attackers.

In this step, you will build dbt mart models that use window functions and CTEs to detect brute-force login patterns and suspicious after-hours access. These are the outputs your dashboard and security team will actually consume.

In this step, get ready to:
  • Build a brute-force detection model using window functions to find 5+ login failures per hour.
  • Build an after-hours access model using the FILTER clause and array aggregation.
  • Add dbt tests and validate the full pipeline.
Create the brute-force detection model

A brute-force attack shows up as a burst of failed logins from the same user and IP in a short window. Your model needs to count failures per user, per IP, per hour, then flag any combination with 5 or more.

  • Create a new file called mart_brute_force_attempts.sql inside models/marts/ and paste the first CTE:
with login_failures as (

    select
        username,
        source_ip,
        event_timestamp,
        count(*) over (
            partition by username, source_ip,
                         date_trunc('hour', event_timestamp)
        ) as failures_per_hour
    from {{ ref('stg_security_events') }}
    where event_type = 'LOGIN_FAILURE'

),

What does this code do?

  • The where event_type = 'LOGIN_FAILURE' clause filters the staging view down to only failed login events.
  • count(*) over (partition by ...) is a window function. It counts failures for each unique username, source IP, and hour window without collapsing rows. Every row keeps its original data while gaining a failures_per_hour total.
  • date_trunc('hour', event_timestamp) groups timestamps into one-hour buckets so that rapid-fire failures in the same hour are counted together.
  • Add the second CTE and final query below the trailing comma in the same file:
brute_force_windows as (

    select
        username,
        source_ip,
        date_trunc('hour', event_timestamp) as attack_window,
        failures_per_hour as failure_count,
        min(event_timestamp) as first_attempt,
        max(event_timestamp) as last_attempt
    from login_failures
    where failures_per_hour >= 5
    group by
        username,
        source_ip,
        date_trunc('hour', event_timestamp),
        failures_per_hour

)

select * from brute_force_windows
order by failure_count desc

What does this code do?

  • where failures_per_hour >= 5 keeps only the rows where 5 or more failures happened in the same hour window. This is the brute-force threshold.
  • The group by collapses those rows into one row per detected attack, while min() and max() capture when the attack window started and ended.
  • The final order by failure_count desc puts the worst attacks at the top.

Why a window function instead of just GROUP BY?

A plain GROUP BY would collapse every failure into a single count. You would lose the individual event rows and their timestamps. The window function counts failures across the window while keeping every row intact, so the second CTE can then filter and re-group only the rows that crossed the threshold.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm your mart_brute_force_attempts.sql contains these sections in order:

  • The login_failures CTE with the count(*) over (...) window function, ending with a comma.
  • The brute_force_windows CTE filtering for failures_per_hour >= 5 and grouping by username, source IP, and hour.
  • The final select * from brute_force_windows ordered by failure_count desc.
  • Save mart_brute_force_attempts.sql.
  • Build the model by running this command in your terminal:
dbt run

You should see output showing the mart_brute_force_attempts model created as a table, alongside the existing staging view.

  • Switch to your psql session and query the new table by running:
SELECT * FROM analytics.mart_brute_force_attempts LIMIT 5;

You should see rows with usernames, source IPs, and failure counts of 5 or more.

Not seeing any rows?

  • Check that your where event_type = 'LOGIN_FAILURE' uses uppercase LOGIN_FAILURE. The staging model normalizes event types to uppercase, so a lowercase filter would match nothing.
  • Make sure the threshold is failures_per_hour >= 5, not > 5. A strict greater-than would miss windows with exactly 5 failures.
  • If dbt run itself failed, check for a missing comma after the first CTE's closing parenthesis.

Help me debug my mart_brute_force_attempts.sql model

Create the after-hours access model

After-hours access is another threat signal. Events outside business hours (before 6 AM, after 8 PM, or on weekends) could indicate unauthorized activity, especially when they involve high-risk actions.

  • Create a new file called mart_after_hours_access.sql inside models/marts/ and paste the CTE that filters for after-hours events:
with after_hours_events as (

    select
        username,
        event_type,
        source_ip,
        department,
        event_timestamp,
        event_hour,
        day_of_week,
        risk_level
    from {{ ref('stg_security_events') }}
    where event_hour < 6
       or event_hour > 20
       or day_of_week in (0, 6)

)

What does this code do?

  • The CTE pulls every column from the staging view but filters to only events that happen before 6 AM (event_hour < 6), after 8 PM (event_hour > 20), or on weekends (day_of_week in (0, 6)).
  • In PostgreSQL, extract(dow ...) returns 0 for Sunday and 6 for Saturday, so in (0, 6) catches both weekend days.
  • Add the aggregation query below the closing parenthesis in the same file, with a blank line between them:
select
    username,
    department,
    count(*) as after_hours_events,
    count(distinct source_ip) as unique_ips,
    count(*) filter (
        where risk_level in ('high', 'critical')
    ) as high_risk_count,
    min(event_timestamp) as first_event,
    max(event_timestamp) as last_event,
    array_agg(distinct event_type) as event_types
from after_hours_events
group by username, department
order by high_risk_count desc, after_hours_events desc

What does this code do?

  • count(*) as after_hours_events totals all off-hours events per user and department.
  • count(*) filter (where risk_level in ('high', 'critical')) uses PostgreSQL's FILTER clause to count only the high-severity events without a separate subquery.
  • array_agg(distinct event_type) collects every distinct event type into a single array column. This lets you see at a glance whether a user was doing logins, privilege escalations, file accesses, or all of the above.
  • Results are sorted by high_risk_count first, so the users with the most dangerous after-hours activity float to the top.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm your mart_after_hours_access.sql contains these sections in order:

  • The after_hours_events CTE selecting from the staging view with three where conditions (before 6 AM, after 8 PM, weekends).
  • The final select with count(*), count(*) filter (...), array_agg(distinct event_type), grouped by username and department, ordered by high_risk_count desc.
  • Save mart_after_hours_access.sql.
  • Build both mart models by running this command:
dbt run

You should see output showing both mart_brute_force_attempts and mart_after_hours_access created as tables, plus the staging view.

  • Switch to your psql session and query the after-hours model by running:
SELECT username, department, after_hours_events, high_risk_count
FROM analytics.mart_after_hours_access
LIMIT 10;

You should see usernames sorted by high-risk event count, with columns for their department, total after-hours events, and how many were high or critical severity.

Seeing a relation does not exist error?

  • Make sure you are querying from the analytics schema (analytics.mart_after_hours_access), not the public schema.
  • If dbt run showed an error for this model, check that the where clause uses or between the three conditions, not and. Using and would require all three to be true simultaneously, which is impossible.
  • Verify you connected to the security_logs database in psql. Run \conninfo to check.

Help me debug my after-hours access dbt model

Add dbt tests and validate the pipeline

The mart tables are built, but you have no automated checks on them yet. Adding dbt tests ensures that every future pipeline run catches problems like null usernames or missing failure counts before they reach your dashboard.

  • Create a new file called schema.yml inside models/marts/ and paste this test configuration:
version: 2

models:
  - name: mart_brute_force_attempts
    description: "Detected brute force login attempts with 5+ failures per hour from the same user and IP"
    columns:
      - name: username
        tests:
          - not_null
      - name: failure_count
        tests:
          - not_null

  - name: mart_after_hours_access
    description: "Users with security events outside business hours or on weekends"
    columns:
      - name: username
        tests:
          - not_null
      - name: after_hours_events
        tests:
          - not_null

What does this file do?

  • Each not_null test tells dbt to query the column and fail if any row contains a null value.
  • The tests run against both mart models: username and failure_count on the brute-force model, and username and after_hours_events on the after-hours model.
  • Combined with the staging tests from the previous step, dbt test now validates data quality across your entire pipeline in one command.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before continuing.

ⓧ I'd like to double check the full code

version: 2

models:
  - name: mart_brute_force_attempts
    description: "Detected brute force login attempts with 5+ failures per hour from the same user and IP"
    columns:
      - name: username
        tests:
          - not_null
      - name: failure_count
        tests:
          - not_null

  - name: mart_after_hours_access
    description: "Users with security events outside business hours or on weekends"
    columns:
      - name: username
        tests:
          - not_null
      - name: after_hours_events
        tests:
          - not_null
  • Save schema.yml.

Before you run the tests, do you think all of them will pass? Your staging model already strips nulls from usernames, and the mart models only produce rows that meet their filter thresholds.

  • Run the full test suite by running this command:
dbt test

All tests pass. You should see green checkmarks for both the staging tests (from the previous step) and the new mart tests.

Tests failing?

  • If a not_null test fails on failure_count, check that schema.yml references the column name failure_count (not failures_per_hour). The second CTE renames it.
  • If a test fails with a compilation error, check the indentation in your YAML. Each tests: key must be nested under its name: key with consistent spacing.
  • Run dbt run before dbt test if you have not rebuilt the models since your last edit. Tests run against the tables as they were last built.

Help me debug my failing dbt tests on the mart models

You now have a complete transformation layer. Raw data flows through the staging model to get cleaned, then through two mart models that surface real threat patterns. Your pipeline detects brute-force attacks and after-hours anomalies automatically every time you run it.

The mart tables hold the threat intelligence, but nobody on your team is going to run SQL queries to check for attacks every morning. Next up, you will build a Streamlit dashboard that makes these insights visual and interactive.

Build the Streamlit Threat Dashboard

Your dbt pipeline is producing clean, actionable threat data. The mart tables hold brute-force detection results and after-hours access patterns, all sitting inside PostgreSQL ready to query.

The problem is that nobody on a security team is going to open psql and run SQL every time they want to check for attacks. In this step, you will build a Streamlit dashboard that pulls data directly from your dbt-created tables and presents it as KPI cards, sortable tables, and distribution charts.

In this step, get ready to:
  • Set up a Streamlit app with cached PostgreSQL connections.
  • Display KPI metrics for total events, brute-force incidents, and after-hours users.
  • Add threat detail tables and event distribution charts.
Create the dashboard file with KPI metrics

The dashboard needs two things before it can display any data: a connection to PostgreSQL and a way to cache query results so the page does not re-query the database on every interaction.

  • Create a new file called dashboard.py in the security-etl root directory (the same folder that contains generate_logs.py).
  • Add the imports and connection helpers by pasting this code:
import streamlit as st
import pandas as pd
from sqlalchemy import create_engine

# Connection string for local PostgreSQL
DB_URL = "postgresql+psycopg2://your-username:your-password@localhost:5432/security_logs"
ANALYTICS_SCHEMA = "analytics"


# Cache the engine so it's shared across Streamlit reruns
@st.cache_resource
def get_engine():
    return create_engine(DB_URL)


# Cache query results for 60 seconds
@st.cache_data(ttl=60)
def load_data(query):
    engine = get_engine()
    return pd.read_sql(query, engine)
  • Replace your-username and your-password with the same PostgreSQL credentials you used in generate_logs.py.

What does this code do?

DB_URL and ANALYTICS_SCHEMA store the database connection string and schema name so every query points at the right place.

@st.cache_resource on get_engine() creates the SQLAlchemy engine once and reuses it across Streamlit reruns, instead of opening a new connection every time you interact with the page. @st.cache_data(ttl=60) on load_data() caches query results for 60 seconds so the dashboard does not re-query PostgreSQL on every click.

  • Below the load_data function, add the page configuration and KPI metrics section:
# Page configuration and title
st.set_page_config(page_title="Security Threat Dashboard", layout="wide")
st.title("Security Threat Detection Dashboard")

# Three KPI columns across the top
col1, col2, col3 = st.columns(3)

total_events = load_data(
    f"SELECT count(*) as total FROM {ANALYTICS_SCHEMA}.stg_security_events"
)
brute_force_count = load_data(
    f"SELECT count(*) as total FROM {ANALYTICS_SCHEMA}.mart_brute_force_attempts"
)
after_hours_count = load_data(
    f"SELECT count(*) as total FROM {ANALYTICS_SCHEMA}.mart_after_hours_access"
)

# Display each metric in its column
col1.metric("Total Cleaned Events", f"{total_events['total'].iloc[0]:,}")
col2.metric("Brute Force Incidents", brute_force_count["total"].iloc[0])
col3.metric("After-Hours Users", after_hours_count["total"].iloc[0])

What does this code do?

st.set_page_config() sets the browser tab title and switches to a wide layout so the dashboard uses the full screen width.

st.columns(3) creates a three-column row. Each column runs a count query against one of your dbt tables and displays the result as a st.metric() card. The total events count uses Python's :, format specifier to add comma separators for readability.

  • Save dashboard.py.
  • Launch the dashboard by running this command in your terminal:
streamlit run dashboard.py

What happens when you run this?

Streamlit starts a local web server and opens your default browser to http://localhost:8501. Your terminal is now running the Streamlit process. Do not close this terminal window.

When you edit and save dashboard.py later, Streamlit detects the change and offers to rerun the app. Click Rerun (or enable Always rerun) to see your updates.

You should see three KPI metric cards at the top of the page: Total Cleaned Events (showing 5,000), Brute Force Incidents, and After-Hours Users.

Dashboard not loading?

  • If you see a connection error, check that DB_URL in dashboard.py matches the credentials you used in generate_logs.py.
  • Make sure your PostgreSQL server is running. The dashboard queries the same security_logs database you created in Step 1.
  • If the browser did not open automatically, navigate to http://localhost:8501 manually.

Why isn't my Streamlit dashboard connecting to PostgreSQL?

Add threat tables and distribution charts

KPI cards give the high-level numbers, but a security analyst needs to see the actual records. The next section adds sortable data tables for both threat types and bar charts showing when events cluster.

  • In dashboard.py, add the threat detail tables below the KPI metrics section:
st.divider()

# Brute force attempts table
st.subheader("Brute Force Login Attempts")
bf_data = load_data(f"""
    SELECT username, source_ip, failure_count,
           attack_window, first_attempt, last_attempt
    FROM {ANALYTICS_SCHEMA}.mart_brute_force_attempts
    ORDER BY failure_count DESC
    LIMIT 20
""")
st.dataframe(bf_data, use_container_width=True)

# After-hours access table
st.subheader("After-Hours Access Activity")
ah_data = load_data(f"""
    SELECT username, department, after_hours_events,
           high_risk_count, unique_ips, first_event, last_event
    FROM {ANALYTICS_SCHEMA}.mart_after_hours_access
    ORDER BY high_risk_count DESC
    LIMIT 20
""")
st.dataframe(ah_data, use_container_width=True)

What does this code do?

Each section runs a SQL query against one of your dbt mart tables and displays the top 20 results in a sortable, scrollable st.dataframe() table. The brute-force table shows which users and IPs had 5+ login failures in a single hour. The after-hours table shows who is active outside business hours and how many high-risk events they triggered.

  • Save dashboard.py and switch to your browser.
  • Click Rerun when Streamlit prompts you.

Below the KPI cards, you should now see two data tables. The brute-force table shows usernames, source IPs, and failure counts. The after-hours table shows departments, event counts, and high-risk totals.

Tables showing up empty?

  • Make sure you ran dbt run in Step 4 to build the mart tables. The dashboard queries analytics.mart_brute_force_attempts and analytics.mart_after_hours_access directly.
  • Check that ANALYTICS_SCHEMA is set to "analytics" at the top of your file.

Why are my Streamlit data tables empty when the dbt models should have data?

  • Below the after-hours table, add the distribution charts:
st.divider()

chart_col1, chart_col2 = st.columns(2)

# Hourly event distribution chart
with chart_col1:
    st.subheader("Events by Hour of Day")
    hourly = load_data(f"""
        SELECT event_hour, count(*) as event_count
        FROM {ANALYTICS_SCHEMA}.stg_security_events
        GROUP BY event_hour
        ORDER BY event_hour
    """)
    st.bar_chart(hourly.set_index("event_hour"))

# Risk level distribution chart
with chart_col2:
    st.subheader("Events by Risk Level")
    risk = load_data(f"""
        SELECT risk_level, count(*) as event_count
        FROM {ANALYTICS_SCHEMA}.stg_security_events
        WHERE risk_level != 'unknown'
        GROUP BY risk_level
        ORDER BY event_count DESC
    """)
    st.bar_chart(risk.set_index("risk_level"))

What does this code do?

st.columns(2) creates a two-column layout. The left chart queries the staging view for event counts grouped by event_hour, showing when security events cluster throughout the day. The right chart groups by risk_level (filtering out unknown) to show how events break down across severity tiers.

Both queries hit the stg_security_events view (the cleaned data), not the raw table. That means every chart reflects the normalized, validated data your staging model produces.

  • Save dashboard.py and switch to your browser.
  • Click Rerun when Streamlit prompts you.

Before you look at the result, think about what you expect the hourly distribution chart to show. Should security events be evenly spread across all 24 hours, or clustered in certain windows?

Scroll down past the data tables. You should see two side-by-side bar charts. The Events by Hour of Day chart shows event counts across all 24 hours. The Events by Risk Level chart breaks down events by severity.

Charts not rendering?

  • Make sure the chart code is inside the with chart_col1: and with chart_col2: blocks. Indentation matters in Python.
  • Check that the queries reference stg_security_events (the staging view), not raw_security_events. The staging view has the event_hour column that the chart needs.

Why are my Streamlit bar charts not showing up?

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboard.py and that the dashboard is running in your browser.

ⓧ I'd like to double check the full code

Your dashboard.py should contain four sections in this order. Scroll up and confirm each one is present:

  • Imports, DB_URL, ANALYTICS_SCHEMA, and the two cached helper functions (get_engine and load_data).
  • Page config, title, and three st.metric() KPI cards in a three-column layout.
  • Two st.dataframe() tables for brute-force attempts and after-hours access, each limited to 20 rows.
  • Two st.bar_chart() charts in a two-column layout for hourly distribution and risk level distribution.

You now have a working security dashboard that reads directly from your dbt-powered analytics layer. Next up, you will take this further with an optional Secret Mission to add interactive filters and drill-down capabilities.

Secret mission

Add Interactive Filters and Drill-Down

Your dashboard shows every event in the dataset at once. A security analyst investigating a specific incident needs to narrow the view to a particular date range and event type. In this mission, you will add sidebar filters that dynamically update every metric, table, and chart on the page.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The security-etl project folder on your Desktop (contains generate_logs.py, dashboard.py, the security_analytics dbt project, and the .venv virtual environment with all installed packages).
  • The security_logs PostgreSQL database (with the raw_security_events table, analytics schema, staging view, and mart tables).
  • The dbt profile file at ~/.dbt/profiles.yml (macOS/Linux) or %USERPROFILE%\.dbt\profiles.yml (Windows).
  • The running Streamlit development server in your terminal.

Keep everything running

No action needed. Choose this if you are still actively building on this pipeline or plan to extend it soon.

  • Your Streamlit dashboard stays running in the terminal. You can stop it anytime with Ctrl+C and restart it later.
  • Your security_logs PostgreSQL database stays on disk. PostgreSQL uses minimal resources when idle.
  • Your security-etl project folder and dbt profile remain in place for future development.

Pause - I'll come back to this later

Stop the running Streamlit process to free up your terminal. Your files and database stay intact so you can pick up where you left off.

  • Switch to the terminal where Streamlit is running and press Ctrl+C to stop the server.
  • When you come back, reactivate your virtual environment and run streamlit run dashboard.py to restart the dashboard.

Delete - I don't want to use this again

Remove all project files, the PostgreSQL database, and the dbt profile to start completely fresh.

  • Switch to the terminal where Streamlit is running and press Ctrl+C to stop the server.
  • Drop the PostgreSQL database by opening a new psql session connected to the default postgres database and running the command below.
DROP DATABASE security_logs;
  • Delete the project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf security-etl

On Windows, run:

cd ~\Desktop
rmdir /s /q security-etl
  • Delete the dbt profile. On macOS/Linux, run:
rm ~/.dbt/profiles.yml

On Windows, run:

del %USERPROFILE%\.dbt\profiles.yml

Other dbt projects on this machine?

If you have other dbt projects that share the same profiles.yml file, open it in a text editor instead and remove only the security_analytics profile block. Deleting the entire file would break your other projects.

Nice Work!

Nice Work!

You just built a complete security threat detection pipeline from scratch. Raw, messy log data goes in one end, and clean, actionable threat intelligence comes out the other.

What you learned:

  • Built a Python data generator that creates realistic security events with intentional quality issues and bulk-loads them into PostgreSQL via SQLAlchemy, forming the Extract and Load stages of your ELT pipeline.
  • Created a dbt project with staging models that clean and normalize raw data, and mart models that detect brute-force login attempts using window functions and after-hours access anomalies using the FILTER clause and array_agg.
  • Built an interactive Streamlit dashboard with KPI metrics, threat data tables, and distribution charts that pull live from your PostgreSQL analytics schema.
  • Secret Mission: Added date-range filtering and event-type drill-down to turn the dashboard into an interactive investigation tool.

Ready to quiz yourself?