Security Analytics Pipeline with dbt

Build an ELT pipeline that detects brute-force attacks using dbt and PostgreSQL.

Introduction

30 Second Summary

Security teams collect thousands of authentication log entries every day. Raw logs are full of mixed casing, broken timestamps, and duplicates, making it nearly impossible to spot real threats like brute-force attacks hiding in the noise.

In this project, you will build a security analytics pipeline that loads messy authentication data into PostgreSQL, transforms it through dbt's layered models, and uses window functions to automatically detect brute-force login patterns.

What You'll Build

Picture querying a single table and instantly seeing a brute-force attack pattern surface: 20 rapid failed logins from one IP address, each flagged automatically by your pipeline's window function analytics.

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

  • A working ELT pipeline that loads 500+ raw security events with a Python script, then uses dbt to transform chaotic data into clean, standardized tables in PostgreSQL.
  • Mart models powered by window functions and CTEs that automatically detect brute-force login patterns and compute daily failure rate statistics.
  • Automated data quality tests and an interactive data lineage graph that traces every table from raw source through staging to analytics-ready marts.
  • Secret Mission: Containerize the entire pipeline with Docker Compose so it runs on any machine with a single command.

Do I need SQL experience for this project?

This project uses intermediate SQL features like JOINs, window functions, and CTEs. Basic Python knowledge is also needed for the data loading script. You should have PostgreSQL installed and running locally before you begin.

Before We Start

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

Set Up Your dbt Environment

You have Python and PostgreSQL ready to go. The missing piece is dbt, the tool that transforms raw data inside your database into clean, tested, analytics-ready tables using pure SQL.

In this step, you will install dbt's PostgreSQL adapter in an isolated virtual environment, scaffold a new dbt project, and verify that dbt can talk to your local database. Once dbt debug reports all green checks, your foundation is solid.

In this step, get ready to:
  • Create a project folder with a Python virtual environment and install dbt-postgres.
  • Initialize the dbt project and configure the PostgreSQL connection.
  • Clean up scaffold files and verify the connection with dbt debug.
Create a project folder and install dbt-postgres

A Python virtual environment keeps dbt's dependencies isolated from your system Python. Installing dbt-postgres pulls in dbt-core and psycopg2-binary as dependencies, so one install gives you everything.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir security-analytics-pipeline
cd security-analytics-pipeline
  • Create and activate a Python virtual environment.

macOS

  • Create and activate the virtual environment by running these commands:
python3 -m venv venv
source venv/bin/activate

Windows

  • Create and activate the virtual environment by running these commands:
python -m venv venv
venv\Scripts\activate

Your terminal prompt now starts with (venv). This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activation command for your operating system. On macOS, use source venv/bin/activate. On Windows, use venv\Scripts\activate.

If you see a permissions error on Windows, try running PowerShell as Administrator and entering Set-ExecutionPolicy RemoteSigned first, then retry activation.

Help me troubleshoot my virtual environment activation.

  • Install dbt-postgres by running:
python -m pip install dbt-postgres

What does this install?

dbt-postgres is the PostgreSQL adapter for dbt. It automatically installs dbt-core (the dbt engine) and psycopg2-binary (the Python driver for PostgreSQL) as dependencies.

  • Confirm the installation succeeded by running:
dbt --version

You should see version numbers for dbt-core and dbt-postgres printed in your terminal.

Seeing 'command not found'?

This usually means the virtual environment is not active. Check that your prompt shows (venv) at the start. If it does not, re-run the activation command for your OS.

If the install itself failed, scroll up in your terminal to find the error. A common cause is an outdated pip. Upgrade it first with python -m pip install --upgrade pip, then retry the install.

Help me fix this dbt installation issue.

Initialize the dbt project and configure the connection

dbt init scaffolds a new project with all the directories and config files dbt needs. It also creates a profiles.yml connection file so dbt knows how to reach your database.

  • Scaffold the dbt project by running:
dbt init security_analytics

What will dbt init ask me?

dbt init walks you through an interactive setup. It will ask you to choose a database adapter, then prompt for connection details one at a time. Here is what to enter at each prompt:

  • Which database would you like to use? Enter 1 to select postgres.
  • host: Enter localhost.
  • port: Enter 5432.
  • user: Enter postgres (or your PostgreSQL username).
  • pass: Enter your PostgreSQL password. The input will be hidden.
  • dbname: Enter security_analytics.
  • schema: Enter analytics.
  • threads: Enter 4.
  • Confirm the project was created by running:
ls

You should see security_analytics and venv listed in the output.

dbt init also created a profiles.yml file with the connection details you entered. You need to verify this file has the right values for our project.

  • Open the profiles file in VS Code.

macOS

  • Open the file by running:
code ~/.dbt/profiles.yml

Windows

  • Open the file by running:
code $HOME\.dbt\profiles.yml

Using Command Prompt instead of PowerShell?

If you are using Command Prompt, run code %USERPROFILE%\.dbt\profiles.yml instead.

  • Replace the entire contents of profiles.yml with the following configuration:
security_analytics:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      user: postgres
      password: your-password-here
      port: 5432
      dbname: security_analytics
      schema: analytics
      threads: 4

What do these fields mean?

  • target: dev tells dbt to use the dev output configuration by default.
  • schema: analytics is the PostgreSQL schema where dbt will create all your models.
  • threads: 4 lets dbt run up to 4 models in parallel for faster builds.
  • Replace your-password-here with your actual PostgreSQL password.
  • Save profiles.yml.

Not sure what your PostgreSQL password is?

This is the password you set when you first installed PostgreSQL. If you used the default installer, the username is postgres and you chose a password during installation.

Help me find my PostgreSQL password.

✔️ Awesome, I've got everything!

Great. Double check you have saved profiles.yml with your real password in place.

ⓧ I'd like to double check the full code

Here is the full profiles.yml file. Make sure your version matches this exactly (except for the password value):

security_analytics:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      user: postgres
      password: your-password-here
      port: 5432
      dbname: security_analytics
      schema: analytics
      threads: 4
Clean up scaffold files and verify the connection

dbt init generated example models that would interfere with your project. Removing them and updating the project configuration prepares the workspace for the security analytics models you will build next.

  • Enter the dbt project folder by running:
cd security_analytics
  • Delete the example models directory.

macOS

  • Remove the directory by running:
rm -rf models/example

Windows

  • Remove the directory by running:
Remove-Item -Recurse -Force models\example
  • Confirm the example directory is gone by running:
ls models

The models directory should now be empty.

  • Open the project configuration file by running:
code dbt_project.yml
  • Replace the entire contents of dbt_project.yml with the following:
name: security_analytics
version: "1.0.0"

profile: security_analytics

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

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

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

What does this configuration do?

  • profile: security_analytics links this project to the connection you configured in profiles.yml.
  • The models section tells dbt to materialize staging models as views (lightweight SQL references) and mart models as tables (physically stored results for faster reads).
  • clean-targets lists the directories that dbt clean will remove to give you a fresh build.
  • Save dbt_project.yml.

✔️ Awesome, I've got everything!

Great. Double check you have saved dbt_project.yml before moving on.

ⓧ I'd like to double check the full code

Here is the full dbt_project.yml. Make sure yours matches exactly:

name: security_analytics
version: "1.0.0"

profile: security_analytics

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

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

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

Before you run the next command, what do you expect to see if your PostgreSQL connection is configured correctly?

  • Verify the full setup by running:
dbt debug

Every check should print green, ending with All checks passed!. You should see lines confirming that profiles.yml and dbt_project.yml are valid, and that the connection test succeeded.

Seeing a connection error?

If dbt debug shows a connection test error, check these common causes:

  • PostgreSQL is not running. Start it with your system's service manager (e.g. brew services start postgresql on macOS or check the Services app on Windows).
  • The password in profiles.yml does not match your PostgreSQL password. Open the file and correct it.
  • The user field does not match a valid PostgreSQL role. The default role is usually postgres.

Help me troubleshoot the dbt debug connection error.

Your dbt environment is connected and ready. Next up, you will load realistic security event data into PostgreSQL and discover why raw data needs transformation.

Load Raw Security Data and See the Problem

Your dbt environment is connected and ready. Before you can transform anything, you need data to transform.

In this step, you'll write a Python script that loads 500+ authentication events into PostgreSQL. The data is intentionally messy: mixed casing, inconsistent timestamps, null values, and duplicate rows. You'll query the raw data and discover firsthand why it cannot be trusted for analytics.

In this step, get ready to:
  • Create a Python data loader that generates security events with realistic quality issues.
  • Query the raw data and see why inconsistent formatting breaks your analysis.
Create the data loader script

The data loader script connects to PostgreSQL, creates a database and table, then fills it with authentication events that have all the quality problems you'd find in a real production logging system.

  • In VS Code, create a new file called load_security_data.py in your project root folder (the folder that contains the security_analytics directory, not inside the dbt project itself).
  • Start the file with imports and the database connection configuration by pasting this code:
import psycopg2
import random
from datetime import datetime, timedelta

# Connection settings for local PostgreSQL
DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "security_analytics",
    "user": "postgres",
    "password": "your-password-here",
}

# 50 simulated user accounts
USERS = [f"user_{i:03d}" for i in range(1, 51)]
# IP addresses including None and empty string for missing data
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",
    "203.0.113.99",
    None,
    "",
]

What does this code do?

DB_CONFIG holds the connection details for your local PostgreSQL. The password value must match the password you set in profiles.yml during Step 1.

The IPS list includes None and an empty string. These simulate the missing and blank IP addresses you'd find in real authentication logs.

  • Replace your-password-here with your actual PostgreSQL password (the same one you used in profiles.yml).
  • Save load_security_data.py.
  • Confirm the file exists by running:
ls load_security_data.py

You should see load_security_data.py listed in the output.

File not found?

Make sure you created the file in your project root folder (the one that contains the security_analytics directory), not inside the dbt project folder. Run ls and check that you can see both load_security_data.py and the security_analytics/ directory at the same level.

Help me find where to create load_security_data.py relative to my dbt project

  • Next, add the remaining data pools below the IPS list. Copy this code and paste it at the bottom of the file:
# Event types with intentional mixed casing
EVENT_TYPES = [
    "LOGIN",
    "login",
    "Login",
    "LOGOUT",
    "logout",
    "FAILED_LOGIN",
    "failed_login",
    "Failed_Login",
    "PASSWORD_CHANGE",
    "MFA_CHALLENGE",
]
# Status values with inconsistent casing
STATUSES = ["success", "SUCCESS", "failure", "FAILURE", "Success", "Failure"]
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
    "curl/7.88.1",
    "python-requests/2.31.0",
    None,
]

Why is the data intentionally messy?

Look at EVENT_TYPES: the same login event appears as LOGIN, login, and Login. Real authentication logs from different services produce exactly this kind of inconsistency.

The same mixed casing appears in STATUSES. These inconsistencies will break any GROUP BY query that tries to aggregate by event type or status.

  • Now add the function that creates the database. Paste this below the constants:
def create_database():
    # Connect to the default 'postgres' database to create our project database
    conn = psycopg2.connect(
        host=DB_CONFIG["host"],
        port=DB_CONFIG["port"],
        dbname="postgres",
        user=DB_CONFIG["user"],
        password=DB_CONFIG["password"],
    )
    conn.autocommit = True
    cur = conn.cursor()
    # Only create if it doesn't already exist
    cur.execute(
        "SELECT 1 FROM pg_database WHERE datname = %s", (DB_CONFIG["dbname"],)
    )
    if not cur.fetchone():
        cur.execute(f"CREATE DATABASE {DB_CONFIG['dbname']}")
        print(f"Created database: {DB_CONFIG['dbname']}")
    else:
        print(f"Database {DB_CONFIG['dbname']} already exists")
    cur.close()
    conn.close()

What does this function do?

create_database() connects to the default postgres database first (you cannot create a new database while connected to it). It checks whether security_analytics already exists before creating it, so running the script twice is safe.

  • Save load_security_data.py and verify the code so far has no syntax errors by running:
python -c "import load_security_data; print('Import successful')"

You should see Import successful printed in the terminal.

Seeing a syntax error or ModuleNotFoundError?

If you see a SyntaxError, check for missing commas between list items and unmatched brackets or quotes. The error message includes a line number pointing to the problem.

If you see a ModuleNotFoundError for psycopg2, make sure your virtual environment from Step 1 is activated. The psycopg2 library comes bundled with dbt-postgres.

Help me fix my import error in load_security_data.py

  • Add the function that creates the raw data table. Paste this below create_database():
def create_tables(conn):
    cur = conn.cursor()
    cur.execute("CREATE SCHEMA IF NOT EXISTS raw")
    cur.execute("DROP TABLE IF EXISTS raw.auth_events")
    # Store all columns as TEXT to simulate untyped raw ingestion
    cur.execute(
        """
        CREATE TABLE raw.auth_events (
            id SERIAL PRIMARY KEY,
            event_timestamp TEXT,
            user_id TEXT,
            event_type TEXT,
            ip_address TEXT,
            status TEXT,
            user_agent TEXT
        )
        """
    )
    conn.commit()
    cur.close()
    print("Created raw.auth_events table")

Why store everything as TEXT?

Every column is TEXT, even event_timestamp. This simulates how raw data often arrives before any type enforcement. The dbt staging layer you build in the next step handles the type conversions.

  • Now add the event generation function. This is the longest function, so you'll add it in two parts. Paste the first part below create_tables():
def generate_events(num_events=500):
    events = []
    base_time = datetime(2026, 8, 20, 0, 0, 0)

    # Generate random events with mixed timestamp formats
    for _ in range(num_events):
        ts = base_time + timedelta(seconds=random.randint(0, 7 * 24 * 3600))
        if random.random() < 0.3:
            ts_str = ts.strftime("%m/%d/%Y %H:%M:%S")
        elif random.random() < 0.5:
            ts_str = ts.isoformat()
        else:
            ts_str = ts.strftime("%Y-%m-%d %H:%M:%S")

        events.append(
            (
                ts_str,
                random.choice(USERS),
                random.choice(EVENT_TYPES),
                random.choice(IPS),
                random.choice(STATUSES),
                random.choice(USER_AGENTS),
            )
        )

What's happening here?

Each event gets a timestamp in one of three formats: US-style (MM/DD/YYYY), ISO 8601, or standard SQL (YYYY-MM-DD). This mixed-format problem is one of the most common data quality issues in production systems.

  • Save load_security_data.py and re-check the syntax by running:
python -c "import load_security_data; print('Import successful')"

You should see Import successful again. If you get a SyntaxError, check the indentation of the events.append() call. It must be inside the for loop.

  • Now add the second part of generate_events(). Paste this directly below the events.append() block you just added, still inside the function (indented one level):
    # Plant a brute-force attack: 20 rapid failures from one IP
    attacker_ip = "203.0.113.42"
    target_user = "user_001"
    brute_start = base_time + timedelta(days=3, hours=2)
    for i in range(20):
        ts = brute_start + timedelta(seconds=i * 5)
        ts_str = ts.strftime("%Y-%m-%d %H:%M:%S")
        events.append(
            (ts_str, target_user, "FAILED_LOGIN", attacker_ip, "failure", "python-requests/2.31.0")
        )

    # Add 15 exact duplicate rows to simulate dirty data
    for _ in range(15):
        events.append(random.choice(events))

    return events

What is the planted brute-force pattern?

This inserts 20 rapid FAILED_LOGIN events from IP 203.0.113.42, all targeting user_001, spaced 5 seconds apart. This is the exact pattern your mart model will detect in Step 4.

The 15 duplicate rows at the end simulate another common production problem. Your staging model will remove these with deduplication.

  • Finally, add the loading function, the main entry point, and the script runner. Paste this below generate_events():
def load_events(conn, events):
    cur = conn.cursor()
    # Insert all events into the raw table
    cur.executemany(
        """
        INSERT INTO raw.auth_events
            (event_timestamp, user_id, event_type, ip_address, status, user_agent)
        VALUES (%s, %s, %s, %s, %s, %s)
        """,
        events,
    )
    conn.commit()
    cur.close()
    print(f"Loaded {len(events)} events into raw.auth_events")


def main():
    create_database()
    conn = psycopg2.connect(**DB_CONFIG)
    create_tables(conn)
    events = generate_events(500)
    load_events(conn, events)
    conn.close()
    print("Done! Raw security data loaded.")


if __name__ == "__main__":
    main()

What does the main function do?

main() runs the full pipeline in order: create the database, create the table, generate 500 random events (plus the 20 brute-force events and 15 duplicates), and load them all into PostgreSQL.

  • Save load_security_data.py.

✔️ Awesome, I've got everything!

Make sure you've saved the file. Your load_security_data.py should contain 7 chunks: imports, DB_CONFIG, 5 constants, and 5 functions (create_database, create_tables, generate_events, load_events, main).

ⓧ I'd like to double check the full code

Scroll up through your code and confirm each section is present in this order:

  • Imports (psycopg2, random, datetime) and DB_CONFIG dictionary.
  • Five constants: USERS, IPS, EVENT_TYPES, STATUSES, USER_AGENTS.
  • create_database() function that checks for and creates the database.
  • create_tables(conn) function that creates the raw.auth_events table.
  • generate_events() function with the random event loop, brute-force planting, and duplicate insertion.
  • load_events(), main(), and the if __name__ == "__main__" block at the very bottom.
Run the loader and query the raw data

The script is ready. Time to load the data and try querying it directly to see what happens.

  • Make sure your terminal is in the project root folder (where load_security_data.py lives). Run the data loader by running:
python load_security_data.py

You should see four lines of output confirming each stage completed successfully:

  • Created database: security_analytics (or already exists if dbt already created it).
  • Created raw.auth_events table.
  • Loaded 535 events into raw.auth_events (the exact count may vary slightly due to the random duplicate selection).
  • Done! Raw security data loaded.

Seeing a connection error?

If you see a connection refused error, make sure PostgreSQL is running. On macOS, check with brew services list. On Windows, check the Services app for the PostgreSQL service.

If you see an authentication error, double-check that the password in DB_CONFIG matches your actual PostgreSQL password. This should be the same password you used in profiles.yml.

Help me fix my PostgreSQL connection error in load_security_data.py

The data is loaded. Now try to analyze it. Before you run the next query, think about this: if you group login events by event_type, how many distinct groups do you expect to see?

  • Connect to the database by running:
psql -d security_analytics -U postgres
  • Run this query to count events by type:
SELECT event_type, count(*) FROM raw.auth_events GROUP BY event_type ORDER BY count DESC;

The results tell a story. Instead of 5 clean event types, you see something like 10 separate groups.

What should I see?

Your results show LOGIN, login, and Login as three separate groups. The same fragmentation affects FAILED_LOGIN and failed_login. PostgreSQL treats each casing as a distinct value.

This means any analysis built on this raw data would split real login counts across multiple rows. A security dashboard reading these numbers would drastically undercount login failures, potentially missing a brute-force attack entirely.

  • Exit psql by typing:
\q

Why does this matter?

You just experienced the core problem that dbt's staging layer solves. The raw data looks fine at first glance, but the inconsistencies produce silently wrong aggregations. No error, no warning. Just misleading numbers.

In the next step, you'll build a dbt staging model that standardizes all of this in one place: uppercase every event type, parse every timestamp format, replace empty strings with nulls, and deduplicate rows.

You've seen the problem with your own eyes. Mixed casing turns 5 real event types into 10 misleading groups. Next up, you'll build dbt staging models that clean all of this in one SQL transformation.

Build dbt Staging Models to Clean the Data

In the last step, you queried raw.auth_events directly and saw the problem firsthand. "LOGIN", "login", and "Login" showed up as separate groups in your results. Mixed timestamp formats and empty IP addresses made the data unreliable for any real analysis.

dbt staging models solve this. They sit between the raw source and your analytics layer, standardizing everything in one place. In this step, you will build a staging model that uppercases event types, parses timestamps, replaces empty strings with nulls, and removes duplicate rows.

In this step, get ready to:
  • Declare raw.auth_events as a dbt source.
  • Build a staging SQL model that standardizes and deduplicates the raw events.
  • Run dbt run --select staging and verify the data is clean.
Declare the raw data as a dbt source

Before dbt can transform your raw table, it needs to know the table exists. dbt uses source declarations to reference tables that live outside dbt's control. This also enables lineage tracking so you can see exactly where your data comes from.

  • In VS Code, right-click the models folder inside your security_analytics dbt project and select New Folder.
  • Name the folder staging.
  • Right-click the new staging folder and select New File.
  • Name the file __sources.yml.
  • Paste the following YAML into __sources.yml:
version: 2

sources:
  - name: raw
    schema: raw
    description: Raw security event data loaded by the Python script
    tables:
      - name: auth_events
        description: Raw authentication event logs with intentional quality issues

What does this file do?

The sources: block tells dbt about tables that exist outside dbt's control. schema: raw points dbt to the raw schema in your PostgreSQL database.

The name: auth_events entry maps to the actual raw.auth_events table your Python script created. Later, your SQL model will reference this source using {{ source('raw', 'auth_events') }} instead of hardcoding the table name.

  • Save __sources.yml.
  • In the VS Code Explorer sidebar, confirm you see __sources.yml inside models/staging/.
Build the staging model to clean the data

Now for the transformation logic. The staging model reads from the raw source, fixes every quality issue you discovered in Step 2, and outputs clean rows. It uses a series of CTEs (Common Table Expressions) to organize each transformation step.

  • Right-click the staging folder and select New File.
  • Name the file stg_auth_events.sql.
  • Paste the following SQL into stg_auth_events.sql to pull raw data and standardize it:
with source as (

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

),

cleaned as (

    select
        id as event_id,
        case
            when event_timestamp ~ '^\d{2}/\d{2}/\d{4}'
                then to_timestamp(event_timestamp, 'MM/DD/YYYY HH24:MI:SS')
            else event_timestamp::timestamp
        end as event_timestamp,
        lower(trim(user_id)) as user_id,
        upper(trim(event_type)) as event_type,
        case
            when trim(ip_address) = '' then null
            else trim(ip_address)
        end as ip_address,
        upper(trim(status)) as status,
        trim(user_agent) as user_agent
    from source
    where event_timestamp is not null

),

What does this code do?

  • The source CTE pulls all rows from raw.auth_events using dbt's {{ source() }} function instead of a hardcoded table name. This is what connects your model to the source you just declared.
  • The cleaned CTE does the heavy lifting. upper(trim(event_type)) forces all event types to uppercase, so "login", "Login", and "LOGIN" all become "LOGIN". lower(trim(user_id)) standardizes user IDs to lowercase.
  • The case block on event_timestamp uses a regex (~) to detect US-format dates like 08/20/2026 14:30:00 and parses them with to_timestamp(). Everything else is cast directly with ::timestamp.
  • Empty IP address strings are replaced with null so they do not pollute aggregations. Rows with a null event_timestamp are filtered out entirely.
  • Add the following SQL directly below the trailing ), in the same file to remove duplicates and produce the final output:

deduplicated as (

    select distinct on (event_timestamp, user_id, event_type, ip_address)
        event_id,
        event_timestamp,
        user_id,
        event_type,
        ip_address,
        status,
        user_agent
    from cleaned
    order by event_timestamp, user_id, event_type, ip_address, event_id

)

select * from deduplicated

How does the deduplication work?

PostgreSQL's distinct on keeps only the first row for each unique combination of event_timestamp, user_id, event_type, and ip_address. The order by clause determines which duplicate row is kept (the one with the lowest event_id).

This removes the 15 exact duplicate rows your Python loader script planted. The final select * from deduplicated outputs the cleaned, deduplicated result set.

  • Save stg_auth_events.sql.

✔️ Awesome, I've got everything!

Great. Double check you have saved stg_auth_events.sql and move on to the next substep.

ⓧ I'd like to double check the full code

Your stg_auth_events.sql file has three CTEs in this order:

  • source pulls from {{ source('raw', 'auth_events') }}.
  • cleaned standardizes casing with upper() and lower(), parses timestamps with to_timestamp() and ::timestamp, and replaces empty IPs with null.
  • deduplicated uses distinct on to remove duplicate rows.
  • The file ends with select * from deduplicated.

Scroll up and confirm each chunk is present in the correct order.

File not showing in the sidebar?

  • Make sure the file is inside models/staging/, not directly in models/. dbt uses the folder structure to apply materialization settings.
  • Check that the filename is exactly stg_auth_events.sql. A typo in the name means dbt will not find the model.

Still stuck? Why isn't my staging model file appearing in the correct location?

Your dbt_project.yml is already configured

In Step 1, you set up dbt_project.yml with +materialized: view for staging models and +materialized: table for marts. That config is already in place, so when you run dbt next, it will create stg_auth_events as a view in the analytics schema automatically.

Run dbt and verify the clean data

Both files are in place. Time to let dbt compile and execute your staging model against PostgreSQL.

Before you run it, think back to the messy GROUP BY results from Step 2. Do you think the same query on the staged data will still show "LOGIN", "login", and "Login" as separate groups, or will they collapse into one?

  • Make sure your terminal is inside the security_analytics dbt project folder (not the parent folder).
  • Run the staging model by running this command:
dbt run --select staging

What should I see?

dbt prints a START line followed by an OK line for each model. You should see output that includes a line ending in [CREATE VIEW] and a final summary of PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1.

  • Now run the same GROUP BY query you ran in Step 2, but this time against the cleaned staging view. Run this in your SQL client:
SELECT event_type, count(*) FROM analytics.stg_auth_events GROUP BY event_type ORDER BY count DESC;

You should see exactly five clean event types: LOGIN, LOGOUT, FAILED_LOGIN, PASSWORD_CHANGE, and MFA_CHALLENGE. Each appears only once with its correct count. Compare that to the fragmented mess from Step 2.

Why is this a view and not a table?

Your dbt_project.yml configures staging models as views. A view does not store data. It re-runs the cleaning SQL every time you query it. This is ideal for staging because it always reflects the latest raw data without duplicating storage.

Mart models (which you will build next) are configured as tables. They store pre-computed results for fast querying.

Seeing an error from dbt run?

  • If dbt reports a source not found error, check that __sources.yml is saved inside models/staging/ and that schema: raw matches the raw schema in your database.
  • If you see a SQL syntax error, open the compiled SQL at target/run/security_analytics/models/staging/stg_auth_events.sql and compare it to your source file for typos.
  • If dbt debug no longer passes, confirm PostgreSQL is still running.

Need more help? Help me debug my dbt run error for the staging model.

Your raw data is clean. Five consistent event types, no duplicates, and proper timestamps. Next up, you will build mart models that use window functions and CTEs to detect brute-force login patterns hiding in this data.

Build Mart Models with Security Analytics

Your staging layer transformed chaotic raw data into clean, standardized events. Every event type is uppercase, every timestamp is parsed, and duplicates are gone.

Clean data is only valuable when it answers real questions. In this step, you will build the mart layer where security analytics happens. You will write two dbt models that use CTEs, window functions, and FILTER clauses to detect security threats. These are the exact SQL patterns that data engineering interviewers test.

In this step, get ready to:
  • Build a daily login statistics model using CTEs and count(*) filter (where ...) clauses.
  • Build a brute-force detection model using window functions and sliding time windows.
  • Run dbt run and query the results to see a brute-force attack pattern surface automatically.
Create the daily login statistics model

The first mart model computes daily login statistics per user. It filters to login-related events, aggregates success and failure counts, counts unique source IPs, and calculates a failure rate percentage.

  • Create the models/marts/ directory inside your security_analytics project folder by running these commands:
mkdir models/marts
ls models/

You should see both staging and marts directories listed.

  • Create a new file called fct_login_attempts.sql inside the models/marts/ folder. Add the two CTEs that filter and aggregate login events:
with login_events as (

    select *
    from {{ ref('stg_auth_events') }}
    where event_type in ('LOGIN', 'FAILED_LOGIN')

),

user_daily_stats as (

    select
        user_id,
        date_trunc('day', event_timestamp)::date as event_date,
        count(*) as total_attempts,
        count(*) filter (where status = 'SUCCESS') as successful_logins,
        count(*) filter (where status = 'FAILURE') as failed_logins,
        count(distinct ip_address) as unique_ips,
        round(
            count(*) filter (where status = 'FAILURE')::numeric
            / nullif(count(*), 0) * 100,
            2
        ) as failure_rate_pct
    from login_events
    group by user_id, date_trunc('day', event_timestamp)::date

)

What does this code do?

  • The login_events CTE pulls only LOGIN and FAILED_LOGIN events from the staging layer using {{ ref('stg_auth_events') }}. This is how dbt tracks dependencies between models.
  • The count(*) filter (where status = 'SUCCESS') syntax counts only rows matching a condition. It replaces clunky CASE WHEN ... THEN 1 END patterns with a cleaner PostgreSQL feature.
  • nullif(count(*), 0) prevents division-by-zero errors when a user has zero total attempts on a given day.
  • Below the closing ) of the user_daily_stats CTE, leave a blank line and add the final select that computes a surrogate key and outputs the statistics:
select
    md5(user_id || '|' || event_date::text) as login_stat_id,
    user_id,
    event_date,
    total_attempts,
    successful_logins,
    failed_logins,
    unique_ips,
    failure_rate_pct
from user_daily_stats

What does the surrogate key do?

md5(user_id || '|' || event_date::text) generates a unique hash for each user-date combination. This gives the table a deterministic primary key without relying on an auto-incrementing ID. If you rerun the model, the same user-date pair always produces the same key.

✔️ Awesome, I've got everything!

Great. Double check you have saved fct_login_attempts.sql.

ⓧ I'd like to double check the full code

Scroll up and confirm your fct_login_attempts.sql contains both parts in order:

  • The login_events and user_daily_stats CTEs, with the count(*) filter aggregations.
  • The final select with the md5() surrogate key and all eight output columns.
  • Save fct_login_attempts.sql.
  • Build just the login attempts model by running this command:
dbt run --select fct_login_attempts

You should see dbt compile and run the model, creating the analytics.fct_login_attempts table.

  • Verify the table contains real data by running this query in psql or your preferred SQL client:
SELECT user_id, event_date, total_attempts, failed_logins, failure_rate_pct
FROM analytics.fct_login_attempts
ORDER BY failure_rate_pct DESC
LIMIT 5;

You should see daily login statistics per user, with failure rates calculated as percentages. Users with higher failure rates appear at the top.

Seeing a compilation error?

  • Check that the file is saved at models/marts/fct_login_attempts.sql (inside marts, not staging).
  • Make sure you have both the CTE block and the final select in the file. A SQL file with only CTEs and no final query is invalid.
  • Confirm the ref('stg_auth_events') matches the staging model filename exactly (without the .sql extension).

Still stuck? Ask for help.

Create the brute-force detection model

The second mart model detects brute-force login attempts using sliding time windows. For each failed login, it counts how many other failures happened from the same user and IP within the preceding 10 minutes.

  • Create a new file called fct_brute_force_detection.sql inside the models/marts/ folder. Add the failed_logins CTE that applies three window functions to each failed login event:
with failed_logins as (

    select
        event_id,
        event_timestamp,
        user_id,
        ip_address,
        user_agent,
        row_number() over (
            partition by user_id, ip_address
            order by event_timestamp
        ) as attempt_number,
        count(*) over (
            partition by user_id, ip_address
            order by event_timestamp
            range between interval '10 minutes' preceding and current row
        ) as failures_in_window,
        event_timestamp - lag(event_timestamp) over (
            partition by user_id, ip_address
            order by event_timestamp
        ) as time_since_last_attempt
    from {{ ref('stg_auth_events') }}
    where event_type = 'FAILED_LOGIN'
      and status = 'FAILURE'

),

What do the three window functions do?

  • row_number() assigns a sequential number to each failed login per user-IP pair, ordered by time. This tells you whether an event is the 1st, 5th, or 20th attempt.
  • count(*) over (... range between interval '10 minutes' preceding and current row) counts how many failures happened in the 10 minutes before each event. This is the sliding window that detects rapid-fire attempts.
  • lag(event_timestamp) grabs the previous event's timestamp so you can compute the time gap between consecutive attempts. A brute-force attacker's gap is often just seconds.
  • Below the closing ), of the failed_logins CTE, leave a blank line and add the brute_force_candidates CTE and the final select:
brute_force_candidates as (

    select
        *,
        case
            when failures_in_window >= 5 then true
            else false
        end as is_brute_force
    from failed_logins

)

select
    event_id,
    event_timestamp,
    user_id,
    ip_address,
    user_agent,
    attempt_number,
    failures_in_window,
    time_since_last_attempt,
    is_brute_force
from brute_force_candidates
where is_brute_force = true
order by user_id, ip_address, event_timestamp

What does this code do?

  • The brute_force_candidates CTE adds a boolean is_brute_force column. Any row with 5 or more failures in its 10-minute window gets flagged as true.
  • The final select filters to only brute-force events with where is_brute_force = true. The output table contains only the suspicious activity, ordered by user, IP, and time.

Why separate CTEs instead of one big query?

Each CTE has one job: failed_logins computes the window metrics, and brute_force_candidates applies the threshold logic. Splitting logic into named CTEs makes complex queries readable and debuggable. This is the pattern interviewers expect to see.

✔️ Awesome, I've got everything!

Great. Double check you have saved fct_brute_force_detection.sql.

ⓧ I'd like to double check the full code

Scroll up and confirm your fct_brute_force_detection.sql contains both parts in order:

  • The failed_logins CTE with the three window functions (row_number(), count(*) over, and lag()), ending with a trailing comma.
  • The brute_force_candidates CTE and the final select with the where is_brute_force = true filter.
  • Save fct_brute_force_detection.sql.
Run dbt and see the attack pattern

Both mart models are ready. Time to build them and see if the brute-force detection actually catches the attack you planted in the raw data.

Before you run this, think about what you expect to see. The loader script planted 20 rapid failed logins from a single IP targeting one user, each 5 seconds apart. Do you think the sliding 10-minute window will catch that pattern?

  • Build all models (staging views and mart tables) by running this command:
dbt run

You should see dbt compile and run all three models: stg_auth_events (view), fct_login_attempts (table), and fct_brute_force_detection (table).

  • Query the brute-force detection table to see the results:
SELECT * FROM analytics.fct_brute_force_detection LIMIT 10;

The results show the planted brute-force pattern: 20 rapid failed logins from IP 203.0.113.42 targeting user_001, each with is_brute_force = true and failures_in_window >= 5. The time_since_last_attempt column shows 5-second gaps between each attempt, confirming the automated attack pattern.

Query returns zero rows?

  • Make sure you ran dbt run (not dbt run --select fct_login_attempts). The brute-force model needs to be built too.
  • Verify the where clause in your failed_logins CTE checks for event_type = 'FAILED_LOGIN' and status = 'FAILURE'. Both conditions must be uppercase to match the staging layer output.
  • Check that the threshold in brute_force_candidates says failures_in_window >= 5, not a higher number.

Still stuck? Ask for help.

Your mart models are surfacing real security threats from the data. A brute-force attack that was invisible in raw logs is now flagged automatically. Next up, you will add automated data quality tests and generate documentation to prove your pipeline is production-ready.

Add Data Quality Tests and Generate Documentation

In Step 4, you built mart models that detect brute-force attack patterns from clean staging data. Your pipeline transforms messy raw events into actionable security analytics.

Without automated checks, a silent SQL bug could produce wrong counts and nobody would notice. In this step, you will add data quality tests across every layer of your dbt project and generate interactive documentation with a data lineage graph.

In this step, get ready to:
  • Add schema tests for data quality across staging and mart models.
  • Run automated tests and confirm zero failures.
  • Generate interactive documentation with a data lineage graph.
Create schema test files for your models

dbt uses YAML files alongside your SQL models to define data quality expectations. Each test declares a contract your data must satisfy every time the pipeline runs.

  • Create a new file called stg_auth_events.yml inside the models/staging/ directory.
  • Add the following schema tests:
version: 2

models:
  - name: stg_auth_events
    description: Cleaned and standardized authentication events with deduplication
    columns:
      - name: event_id
        description: Unique identifier from the source table
        tests:
          - not_null
      - name: event_timestamp
        description: Parsed and standardized event timestamp
        tests:
          - not_null
      - name: event_type
        description: Uppercased event type
        tests:
          - not_null
          - accepted_values:
              values:
                - LOGIN
                - LOGOUT
                - FAILED_LOGIN
                - PASSWORD_CHANGE
                - MFA_CHALLENGE

What do these tests check?

not_null on event_id and event_timestamp ensures every row has an identifier and a timestamp. If your cleaning logic accidentally dropped these values, the test catches it.

accepted_values on event_type locks the column down to exactly five valid types. Any new value or typo that slips through staging fails the test immediately.

  • Save stg_auth_events.yml.
  • Confirm the staging tests work by running this command:
dbt test --select stg_auth_events

You should see all staging tests pass with zero failures.

Seeing test failures or a compilation error?

  • YAML is indentation-sensitive. Make sure each level uses exactly two spaces, not tabs.
  • Confirm the file is saved inside models/staging/, not the project root or another directory.
  • Check that the model name in the YAML (stg_auth_events) matches the SQL filename exactly.

Why are my dbt staging tests failing?

✔️ Awesome, I've got everything!

Great. Make sure stg_auth_events.yml is saved.

ⓧ I'd like to double check the full code

version: 2

models:
  - name: stg_auth_events
    description: Cleaned and standardized authentication events with deduplication
    columns:
      - name: event_id
        description: Unique identifier from the source table
        tests:
          - not_null
      - name: event_timestamp
        description: Parsed and standardized event timestamp
        tests:
          - not_null
      - name: event_type
        description: Uppercased event type
        tests:
          - not_null
          - accepted_values:
              values:
                - LOGIN
                - LOGOUT
                - FAILED_LOGIN
                - PASSWORD_CHANGE
                - MFA_CHALLENGE

Now add tests for the mart layer. These tests check the analytics tables that downstream consumers rely on.

  • Create a new file called marts.yml inside the models/marts/ directory.
  • Add the following schema tests:
version: 2

models:
  - name: fct_login_attempts
    description: Daily login attempt statistics per user with failure rates
    columns:
      - name: login_stat_id
        description: Surrogate key derived from user_id and event_date
        tests:
          - unique
          - not_null
      - name: failure_rate_pct
        description: Percentage of failed login attempts for the user on that day

  - name: fct_brute_force_detection
    description: Detected brute-force login attempts using 10-minute sliding window analysis
    columns:
      - name: event_id
        description: Event identifier from the staging layer
        tests:
          - not_null

What do these tests check?

unique and not_null on login_stat_id ensures every row in fct_login_attempts has a distinct, non-empty surrogate key. A collision would mean your md5() hash produced duplicates.

not_null on event_id in fct_brute_force_detection ensures every flagged attack event traces back to a real source record.

  • Save marts.yml.

Not sure about the YAML structure?

  • Each model entry starts with - name: at the same indentation level. The second model (fct_brute_force_detection) must line up with the first.
  • Make sure marts.yml is saved in models/marts/, not models/staging/.

Help me fix my dbt mart YAML tests.

✔️ Awesome, I've got everything!

Great. Make sure marts.yml is saved.

ⓧ I'd like to double check the full code

version: 2

models:
  - name: fct_login_attempts
    description: Daily login attempt statistics per user with failure rates
    columns:
      - name: login_stat_id
        description: Surrogate key derived from user_id and event_date
        tests:
          - unique
          - not_null
      - name: failure_rate_pct
        description: Percentage of failed login attempts for the user on that day

  - name: fct_brute_force_detection
    description: Detected brute-force login attempts using 10-minute sliding window analysis
    columns:
      - name: event_id
        description: Event identifier from the staging layer
        tests:
          - not_null
Run your tests and generate documentation

Both schema files are in place. Time to run every test across the full pipeline and see if your data passes inspection.

Before you run this, how many individual tests do you think dbt will execute across both the staging and mart layers?

  • Run all tests across the entire project by running this command:
dbt test

dbt runs 7 tests in total: 3 on stg_auth_events (two not_null and one accepted_values), 2 on fct_login_attempts (unique and not_null), and 1 on fct_brute_force_detection (not_null). Every test should pass with zero failures.

Why does this matter for production?

In a production pipeline, dbt test runs after every dbt run. If any test fails, the pipeline halts and alerts the team. This prevents bad data from reaching dashboards and reports.

Seeing test failures?

  • If an accepted_values test fails, your staging model may not be uppercasing event types correctly. Check that stg_auth_events.sql uses upper(trim(event_type)).
  • If a unique test fails on login_stat_id, check that your fct_login_attempts.sql uses the correct group by clause to avoid duplicate user-date combinations.
  • If you see a compilation error, re-check your YAML indentation. Every nested key needs exactly two spaces.

Why are my dbt tests failing?

Passing tests prove your data is clean. Now generate documentation so anyone on your team can see what your pipeline does and how data flows through it.

  • Generate the documentation catalog by running this command:
dbt docs generate

You should see output confirming the catalog was generated.

  • Serve the documentation site locally by running this command:
dbt docs serve

Your default browser opens automatically to an interactive documentation site at localhost:8080. If it does not open on its own, navigate to http://localhost:8080 manually.

What is the lineage graph?

Click the blue circle icon in the bottom-right corner of the docs site to open the lineage graph. It shows how data flows through your pipeline: raw.auth_events feeds into stg_auth_events, which then feeds into both fct_login_attempts and fct_brute_force_detection.

This visual proof of data lineage is a key part of data governance. It shows exactly where every number in your analytics tables comes from.

Stopping the docs server

Press Ctrl+C in the terminal to stop the documentation server when you are done exploring.

Browser did not open?

  • If port 8080 is already in use, try running dbt docs serve --port 8081 instead.
  • Make sure you ran dbt docs generate first. The serve command needs the generated catalog files.

Help me get dbt docs serve working.

Your pipeline is now tested, documented, and fully traceable. Every layer has automated quality checks, and the lineage graph proves exactly how data flows from raw source through staging to analytics-ready marts. That is a production-grade pipeline.

Secret mission

Containerize the Pipeline with Docker

Your pipeline works on your machine, but production teams need it to run anywhere. Containerize the entire security analytics pipeline with Docker Compose so a single command spins up PostgreSQL, loads data, runs dbt transformations, and passes all tests.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local machine. There are no cloud resources or paid services, so there are no ongoing costs.

Resources you used:

  • security_analytics PostgreSQL database with raw and analytics schemas.
  • ~/.dbt/profiles.yml dbt configuration file.
  • Project folder on the Desktop containing Python scripts, dbt models, and the virtual environment.
  • Docker containers, images, and volumes (if you completed the Secret Mission).

Keep everything running

No action needed. Your project runs locally with no ongoing costs, so everything can stay as-is.

  • Your security_analytics database stays on disk and is available whenever PostgreSQL is running.
  • Your project folder, virtual environment, and dbt configuration remain ready for continued experimentation.
  • If you completed the Secret Mission, your Docker images remain cached locally for fast rebuilds.

Pause - I'll come back to this later

Stop any running processes to free up memory and ports. Your files and database stay on disk for when you return.

  • If dbt docs serve is still running, press Ctrl+C in that terminal to stop the documentation server.
  • If you completed the Secret Mission and Docker containers are running, stop them by running this from your project folder:
docker compose down
  • Your database, project files, and dbt configuration all remain on disk.

Delete - I don't want to use this again

Remove all project resources and start fresh.

  • If you completed the Secret Mission, stop and remove all Docker resources (containers, networks, and the pgdata volume) by running this from your project folder:
docker compose down -v
  • If your virtual environment is still active, deactivate it by running:
deactivate
  • Drop the security_analytics PostgreSQL database by running:
dropdb -U postgres security_analytics
  • Remove the dbt configuration file. On macOS/Linux:
rm ~/.dbt/profiles.yml
  • On Windows:
del $env:USERPROFILE\.dbt\profiles.yml
  • Delete your project folder from the Desktop. On macOS/Linux, run the following command, replacing the folder name with your actual project folder:
rm -rf ~/Desktop/[[PROJECT_FOLDER="your-project-folder-name"]]
  • On Windows, open File Explorer, navigate to your Desktop, right-click the project folder, and select Delete.

Nice Work!

Nice Work!

You just built a production-style security analytics pipeline from scratch using Python, dbt, and PostgreSQL. You took chaotic raw authentication logs, transformed them through a layered dbt pipeline, and surfaced a real brute-force attack pattern with advanced SQL analytics.

You've learned how to:

  • Build an ELT pipeline that extracts and loads 500+ security events into PostgreSQL with a Python data loader, then transforms the messy raw data through dbt staging models that standardize casing, parse mixed timestamp formats, handle nulls, and deduplicate rows.
  • Write mart models using window functions (sliding 10-minute time windows), CTEs, and FILTER clauses to detect brute-force login patterns and compute daily login failure statistics.
  • Enforce data quality with automated dbt tests covering uniqueness, not-null, and accepted value constraints, then generate interactive documentation and a data lineage graph showing how data flows from raw source through staging to analytics-ready marts.
  • Secret Mission: Containerized the full pipeline with Docker Compose so it runs anywhere with a single command, the same deployment pattern production data engineering teams use.

Ready to quiz yourself?