Security Log ETL Pipeline with NXLog

Build an automated ETL pipeline that collects Windows Event Logs and visualizes them.

Introduction

30 Second Summary

Every login attempt, system error, and configuration change on your computer generates an event log entry. Most of these logs go completely unread, even though they hold the clues to security breaches and unauthorized access.

In this project, you will build a complete ETL pipeline that collects real Windows Event Logs from your own machine, transforms them into structured analytical data, and visualizes security patterns through an interactive dashboard.

What You'll Build

Picture typing one command and watching a browser tab fill with live charts of your machine's security events, showing severity spikes, the noisiest event sources, and a timeline of every login and system change your computer recorded.

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

  • An ETL pipeline in Python that extracts real Windows Event Logs collected by NXLog CE, classifies events by severity, and loads clean records into a DuckDB analytical database.
  • An automated file-watching layer using watchdog that detects new log data and triggers the pipeline without any manual intervention.
  • An interactive Streamlit dashboard with Plotly charts for exploring event timelines, severity breakdowns, and top event sources from your own machine's logs.
  • Secret Mission: Add anomaly frequency detection that flags events occurring at unusual frequencies and highlights them on your dashboard.

Are there any prerequisites?

This project runs on a Windows machine because NXLog CE collects Windows Event Logs. You should be comfortable with intermediate Python (functions, file I/O, dictionaries) and have VS Code and Python 3.10+ already installed.

Before We Start

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

Set Up NXLog CE and Your Python Environment

Your ETL pipeline needs two foundations before it can process a single log event: a source of real data and a Python environment to process it.

NXLog Community Edition is a lightweight log collector that reads Windows Event Logs and outputs them as structured JSON files. In this step, you'll install NXLog CE and set up a Python environment with the packages that power the rest of the pipeline.

In this step, get ready to:
  • Install NXLog Community Edition on your Windows machine.
  • Create a project directory with a Python virtual environment and required packages.
  • Verify that NXLog CE and your Python packages are working.
Download and install NXLog Community Edition

NXLog Community Edition (CE) is the free, open-source log collector that will feed real Windows Event Logs into your pipeline. You need to download the installer from the NXLog website and run it on your machine.

  • Open your browser and navigate to the NXLog CE downloads page.
  • Find the Windows x86-64 row and download the nxlog-ce-3.2.2329.msi package.
  • Double-click the downloaded .msi file to launch the installer.
  • Windows may show a User Account Control prompt asking for permission. Click Yes to allow the installer to run.
  • Accept the license agreement.
  • Use the default installation directory (C:\Program Files\nxlog).
  • Click through the remaining prompts to complete the installation.

What is NXLog CE?

NXLog Community Edition is a free, open-source log collection agent. It can read logs from many sources, including Windows Event Logs, and forward them in different formats.

In this project, you'll configure it to collect Security, System, and Application channel events and output them as JSON. Your Python ETL pipeline will then consume that JSON output.

Create your project directory and Python environment

With NXLog CE installed, you now need a Python project directory and the packages your pipeline depends on.

  • Press the Windows key to open your search bar.
  • Type PowerShell and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir security-log-etl
cd security-log-etl

Your terminal prompt should now show security-log-etl as the current directory.

  • Set up a Python virtual environment and activate it by running these commands:
python -m venv venv
venv\Scripts\Activate

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

Seeing a script execution error?

PowerShell may block the activation script with an error containing the phrase "running scripts is disabled on this system." This is a common default on Windows. Fix it by running this command:

Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

After running the command, try venv\Scripts\Activate again. Your prompt should then show (venv).

Help me fix my PowerShell execution policy so I can activate my Python virtual environment.

  • Install the required Python packages by running:
pip install duckdb==1.5.5 streamlit==1.62.0 watchdog==6.0.0 plotly==7.0.0

You should see download progress bars followed by a message confirming the packages were successfully installed.

What are these packages?

  • DuckDB is an analytical database that stores and queries your processed log data.
  • Streamlit is a Python framework for building interactive web dashboards.
  • watchdog monitors a directory for file changes and triggers actions automatically.
  • Plotly creates interactive charts you can zoom, hover, and filter.

Package installation failed?

Make sure your virtual environment is active. Your prompt should start with (venv). If it does not, run venv\Scripts\Activate first, then retry the pip install command.

Help me troubleshoot my pip install failure for the security log ETL project.

Verify your setup

Both tools are installed. Before running the verification commands, think about what you expect to see if everything installed correctly.

  • Check that NXLog CE is installed by running this command:
& "C:\Program Files\nxlog\nxlog.exe" -v

You should see a version number printed, confirming NXLog CE is installed at the expected location.

Command not recognized or file not found?

Make sure the quotes around the path are included. The space in Program Files requires them. Also verify you used the default installation directory during setup.

If NXLog is installed in a different location, right-click nxlog.exe in File Explorer, click Properties, and use the path shown in the Location field.

Help me find and verify my NXLog CE installation on Windows.

Now check that DuckDB installed correctly. What version number do you expect to see?

  • Verify the DuckDB package by running:
python -c "import duckdb; print(duckdb.__version__)"

You should see 1.5.5 printed. This confirms DuckDB is installed and importable in your virtual environment.

Seeing ModuleNotFoundError?

This means Python cannot find the duckdb package. Check that your virtual environment is active by looking for (venv) in your prompt. If it is missing, activate it with venv\Scripts\Activate and try the command again.

Help me figure out why Python can't find duckdb in my virtual environment.

Your tools are installed and verified. Next up, you'll configure NXLog CE to start collecting real Windows Event Logs and outputting them as JSON.

Configure NXLog CE to Collect Windows Event Logs as JSON

Your NXLog CE installation is ready, but it does not know what to collect or where to send it. NXLog CE is the upstream data source for your entire ETL pipeline. Without it producing structured output, there is no data to extract, transform, or load.

In this step you will configure NXLog CE to collect real Windows Event Logs from three channels (Security, System, and Application), convert every event to JSON, and write the output to a file your Python ETL can read. By the end, your machine will be continuously generating structured log data.

In this step, get ready to:
  • Create the output directory and open the NXLog CE configuration file.
  • Write the NXLog configuration that collects events from three Windows Event Log channels as JSON.
  • Restart the NXLog service and verify JSON events are being written to the log file.
Create the output directory and open the configuration file

NXLog CE writes its output to a file on disk. The output directory must exist before NXLog starts writing, otherwise the service fails silently.

  • Open an Administrator terminal by pressing the Windows key, typing cmd, right-clicking Command Prompt, and selecting Run as administrator.
  • Create the output directory by running this command:
mkdir C:\logs
  • Confirm the directory exists by running:
dir C:\logs

You should see an empty directory listing with no files yet.

Why an Administrator terminal?

The NXLog CE configuration file lives inside C:\Program Files\nxlog, which is a protected directory on Windows. You need Administrator privileges to edit files there and to start or stop the NXLog service.

  • Open the NXLog CE configuration file in VS Code as Administrator by running:
code "C:\Program Files\nxlog\conf\nxlog.conf"

VS Code not opening as Administrator?

If VS Code opens without write access, close it and launch VS Code itself as Administrator first. Press the Windows key, type VS Code, right-click, and select Run as administrator. Then open the file with File > Open File and navigate to C:\Program Files\nxlog\conf\nxlog.conf.

Help me open the NXLog config file as Administrator

Write the NXLog configuration for three event log channels

The default nxlog.conf file contains sample configuration that does not do anything useful yet. You will replace it entirely with a configuration that collects events from the Security, System, and Application channels.

  • Select all the content in nxlog.conf by pressing Ctrl+A, then delete it.
  • Paste in the following configuration that defines constants, the JSON extension, and the input module:
# NXLog CE configuration for Windows Event Log collection
# Collects Security, System, and Application events as JSON

define      INSTALLDIR    C:\Program Files\nxlog
define      LOGDIR        %INSTALLDIR%\data
define      MYLOGFILE     %LOGDIR%\nxlog.log

LogLevel    INFO
LogFile     %MYLOGFILE%

# Load the JSON extension for converting events
<Extension json>
    Module          xm_json
</Extension>

What does this code do?

The define lines create reusable constants for the NXLog installation directory and its internal log file. These are not your output logs.

The Extension json block loads the xm_json module, which provides the to_json() procedure you will use in the output section to convert each event into a JSON string.

  • Below the Extension block, paste the input module that collects from three Windows Event Log channels:
# Collect Windows Event Logs from three channels
<Input eventlog>
    Module          im_msvistalog
    ReadFromLast    FALSE
    SavePos         FALSE
    <QueryXML>
        <QueryList>
            <Query Id='0'>
                <Select Path='Application'>*</Select>
                <Select Path='Security'>*</Select>
                <Select Path='System'>*</Select>
            </Query>
        </QueryList>
    </QueryXML>
</Input>

What does this code do?

The im_msvistalog module uses the Windows Event Log API to collect events. The QueryXML block uses standard Windows XPath syntax to select all events from the Application, Security, and System channels.

Setting ReadFromLast and SavePos to FALSE tells NXLog to read all existing events from the beginning, not just new ones. This gives you a large batch of historical data to work with immediately.

  • Below the Input block, paste the output module and route that write JSON events to the log file:
# Write each event as a JSON line to the output file
<Output output_file>
    Module          om_file
    File            'C:\logs\windows_events.log'
    Exec            to_json();
</Output>

# Connect input to output
<Route r1>
    Path            eventlog => output_file
</Route>

What does this code do?

The om_file output module writes each event to C:\logs\windows_events.log. The Exec to_json(); line calls the JSON extension to convert each event's fields into a single-line JSON string before writing it.

The Route block connects the input (eventlog) to the output (output_file), defining the data flow. Every event collected by the input module flows through this route to the output file.

  • Save nxlog.conf by pressing Ctrl+S.

\u2714\uFE0F Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

\u24E7 I'd like to double check the full code

Your complete nxlog.conf should look exactly like this:

# NXLog CE configuration for Windows Event Log collection
# Collects Security, System, and Application events as JSON

define      INSTALLDIR    C:\Program Files\nxlog
define      LOGDIR        %INSTALLDIR%\data
define      MYLOGFILE     %LOGDIR%\nxlog.log

LogLevel    INFO
LogFile     %MYLOGFILE%

# Load the JSON extension for converting events
<Extension json>
    Module          xm_json
</Extension>

# Collect Windows Event Logs from three channels
<Input eventlog>
    Module          im_msvistalog
    ReadFromLast    FALSE
    SavePos         FALSE
    <QueryXML>
        <QueryList>
            <Query Id='0'>
                <Select Path='Application'>*</Select>
                <Select Path='Security'>*</Select>
                <Select Path='System'>*</Select>
            </Query>
        </QueryList>
    </QueryXML>
</Input>

# Write each event as a JSON line to the output file
<Output output_file>
    Module          om_file
    File            'C:\logs\windows_events.log'
    Exec            to_json();
</Output>

# Connect input to output
<Route r1>
    Path            eventlog => output_file
</Route>
Restart the NXLog service and verify JSON output

The NXLog service loaded the old configuration when it started. To pick up your new configuration, you need to restart the service.

  • Switch back to your Administrator terminal.
  • Stop the NXLog service by running:
net stop nxlog
  • Start the service again with the new configuration by running:
net start nxlog

You should see a message confirming the service started successfully.

Why two separate commands?

Running net stop and net start as separate steps makes it clear when the old config has been fully unloaded. If the stop fails (because the service was not running), you can still run the start command safely.

  • Wait 10-15 seconds for NXLog to collect and write events.

Before you check the output file, what do you think each line will look like? Recall that you configured NXLog to convert events using to_json() before writing them.

  • Open the output file to verify events are being written by running:
code C:\logs\windows_events.log

You should see the file filled with JSON-formatted Windows Event Log entries, one per line. Each line contains fields like EventTime, EventID, Severity, SourceName, Channel, Hostname, Message, and EventReceivedTime.

File is empty or does not exist?

  • Make sure the C:\logs directory exists. Run dir C:\logs to check. If it does not exist, create it with mkdir C:\logs and restart NXLog.
  • Check the NXLog internal log for errors by opening C:\Program Files\nxlog\data\nxlog.log in VS Code. Look for lines containing ERROR which indicate a configuration problem.
  • Verify the service is actually running by running sc query nxlog in your Administrator terminal. The state should show RUNNING.

Help me troubleshoot my NXLog configuration

What is NXLog collecting?

With ReadFromLast set to FALSE, NXLog reads all historical events from your machine's Security, System, and Application channels. You could see thousands of events appear immediately.

These are real events from your own machine. Logins, service starts, application errors, security audits. This is the same data that enterprise security teams analyse to detect threats.

Your log collector is live and producing real data. Next up, you will build the Python ETL pipeline that reads this JSON output, transforms the messy fields into a clean schema, and loads everything into a DuckDB analytical database.

Build the ETL Pipeline to Parse and Load Logs into DuckDB

NXLog CE is collecting real Windows Event Logs and writing them as JSON to C:\logs\windows_events.log. That file is growing every second, but raw JSON lines sitting in a folder are not something you can query or analyze.

This step builds the core of any data engineering project: an ETL pipeline. ETL stands for Extract, Transform, Load. You will write a Python script that extracts fields from the JSON, transforms them into a clean schema, and loads them into a DuckDB analytical database you can query instantly.

In this step, get ready to:
  • Extract fields from NXLog's JSON log output line by line.
  • Transform event data with timestamp normalization and severity classification.
  • Load the cleaned records into a DuckDB analytical database.
Set up etl.py with imports and the severity map

NXLog outputs a raw Severity string for each event, like INFO, WARNING, or AUDIT_FAILURE. Different event channels use different strings, so you need a mapping to normalize them into consistent categories your dashboard can filter on.

  • Create a new file called etl.py in your project directory by adding the following code:
import json
import os
from datetime import datetime

import duckdb

# File paths for the NXLog log file, DuckDB database, and ETL state tracker
LOG_FILE = r"C:\logs\windows_events.log"
DB_FILE = "security_logs.db"
STATE_FILE = "etl_state.txt"

# Map raw NXLog severity strings to standardized categories
SEVERITY_MAP = {
    "CRITICAL": "Critical",
    "ERROR": "High",
    "WARNING": "Medium",
    "INFO": "Low",
    "DEBUG": "Info",
    "AUDIT_SUCCESS": "Info",
    "AUDIT_FAILURE": "High",
}


def get_last_position():
    # Read the byte offset from the state file to know where to resume
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            return int(f.read().strip())
    return 0


def save_position(position):
    # Save the current byte offset so the next run picks up where this one left off
    with open(STATE_FILE, "w") as f:
        f.write(str(position))

What does this code do?

SEVERITY_MAP translates raw NXLog severity strings into five consistent categories: Critical, High, Medium, Low, and Info. For example, AUDIT_FAILURE maps to High because a failed audit is a security concern.

get_last_position() and save_position() track a byte offset in etl_state.txt. This tells the ETL where it left off, so the next run only processes new lines instead of re-reading the entire file.

  • Save etl.py.
  • Confirm the file exists by running:
ls etl.py
  • You should see etl.py listed in the output.

Don't see etl.py?

  • Make sure you saved the file in your project directory, not somewhere else on your machine.
  • Check that your terminal is in the same directory as your project. Run pwd to confirm.

Help me find where my etl.py file was saved

Add the transform and extract functions

NXLog's JSON output has inconsistent fields across different event types. Some events have a Severity field, others do not. Timestamps include fractional seconds that need stripping. This is exactly the kind of data quality challenge you face in production data engineering.

  • In etl.py, add the two transform helper functions below the save_position() function:
def classify_severity(raw_severity):
    # Return a standardized severity label using the SEVERITY_MAP
    if raw_severity is None:
        return "Info"
    return SEVERITY_MAP.get(raw_severity.upper(), "Info")


def parse_event_time(raw_time):
    # Strip fractional seconds and convert to a Python datetime
    if raw_time is None:
        return None
    try:
        clean = raw_time.split(".")[0]
        return datetime.fromisoformat(clean)
    except (ValueError, AttributeError):
        return None

What do these functions do?

classify_severity() takes the raw string from NXLog (like WARNING or AUDIT_FAILURE) and returns a clean category. If the field is missing entirely, it defaults to Info instead of crashing.

parse_event_time() strips the fractional seconds from NXLog's timestamp format (e.g. 2026-08-27T10:15:30.123 becomes 2026-08-27T10:15:30) and converts it to a Python datetime object that DuckDB can store as a TIMESTAMP.

Now you need the extract function itself. This is the function that opens the NXLog log file, reads each JSON line, and builds a list of cleaned event dictionaries.

  • Still in etl.py, add the extract_events() function below parse_event_time():
def extract_events(log_file, start_position=0):
    events = []
    with open(log_file, "r", encoding="utf-8", errors="replace") as f:
        # Jump to the byte offset where the last run stopped
        f.seek(start_position)
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                raw = json.loads(line)
            except json.JSONDecodeError:
                # Skip lines that are not valid JSON
                continue

            event_time = parse_event_time(raw.get("EventTime"))
            if event_time is None:
                continue

            # Build a cleaned dictionary with defaults for missing fields
            events.append({
                "event_time": event_time,
                "event_id": raw.get("EventID"),
                "severity": classify_severity(raw.get("Severity")),
                "source_name": raw.get("SourceName", "Unknown"),
                "channel": raw.get("Channel", "Unknown"),
                "hostname": raw.get("Hostname", "Unknown"),
                "message": (raw.get("Message") or "")[:500],
                "received_time": parse_event_time(
                    raw.get("EventReceivedTime")
                ),
            })
        new_position = f.tell()
    return events, new_position

What does extract_events do?

  • f.seek(start_position) jumps to the byte offset from the last ETL run so only new lines are processed.
  • Each line is parsed with json.loads(). Malformed lines are skipped instead of crashing the pipeline.
  • raw.get("SourceName", "Unknown") provides safe defaults when NXLog events are missing fields.
  • The message is truncated to 500 characters to keep the database lean.
  • f.tell() captures the new byte position so the next run resumes exactly here.
  • Save etl.py.
  • Test that the extract function can read your NXLog log file by running:
python -c "from etl import extract_events, LOG_FILE; events, pos = extract_events(LOG_FILE); print(f'Extracted {len(events)} events from {pos} bytes')"

Before you run this, think about what you expect. NXLog has been writing events since you started the service. Do you think the count will be in the dozens, hundreds, or thousands?

You should see output like Extracted 347 events from 289450 bytes. Your exact numbers will differ depending on how long NXLog has been running and how active your system is.

Seeing 0 events extracted?

  • Confirm that C:\logs\windows_events.log exists and is not empty. Open it in VS Code to verify it contains JSON lines.
  • Make sure the NXLog service is running. In an Administrator terminal, run net start nxlog and wait 10-15 seconds for events to appear.
  • Check that the LOG_FILE path in your code matches exactly: r"C:\logs\windows_events.log". The r prefix before the string prevents Python from interpreting backslashes as escape characters.

Why is my extract_events function returning 0 events from NXLog logs?

Add the load function and run the full ETL

The extract step pulls raw data out. The transform step cleans it. Now you need the Load step to write the cleaned records into DuckDB so you can run SQL queries against your security events.

  • In etl.py, add the first part of the load_events() function below extract_events(). This creates the database table and handles the empty-events case:
def load_events(events, db_file=DB_FILE):
    con = duckdb.connect(db_file)
    # Create the events table if it does not already exist
    con.sql("""
        CREATE TABLE IF NOT EXISTS events (
            event_time TIMESTAMP,
            event_id INTEGER,
            severity VARCHAR,
            source_name VARCHAR,
            channel VARCHAR,
            hostname VARCHAR,
            message VARCHAR,
            received_time TIMESTAMP
        )
    """)

    if not events:
        print("No new events to load.")
        con.close()
        return 0

What does this code do?

duckdb.connect(db_file) opens (or creates) a persistent database file called security_logs.db. Unlike an in-memory database, this file survives between runs so your data is always available.

CREATE TABLE IF NOT EXISTS creates the table on the first run and safely skips it on later runs. The eight columns match the fields you extracted: timestamps, event metadata, and the truncated message.

The function is not complete yet. It can create the table and handle an empty input, but it cannot insert any records. You need to add the insertion logic.

  • Still inside the load_events() function, add the insertion logic, the run_etl() orchestration function, and the main entry point below the return 0 line:
    # Insert all cleaned events into DuckDB in one batch
    con.executemany(
        """INSERT INTO events VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
        [
            (
                e["event_time"],
                e["event_id"],
                e["severity"],
                e["source_name"],
                e["channel"],
                e["hostname"],
                e["message"],
                e["received_time"],
            )
            for e in events
        ],
    )
    count = len(events)
    print(f"Loaded {count} events into {db_file}.")
    con.close()
    return count


def run_etl():
    # Orchestrate the full Extract-Transform-Load pipeline
    start_pos = get_last_position()
    events, new_pos = extract_events(LOG_FILE, start_pos)
    loaded = load_events(events)
    save_position(new_pos)
    return loaded


if __name__ == "__main__":
    run_etl()

What does this code do?

  • con.executemany() inserts all extracted events into DuckDB in a single batch. This is much faster than inserting one row at a time.
  • run_etl() is the main entry point that ties all three ETL steps together. It reads the last position, extracts new events, loads them into DuckDB, then saves the new position.
  • The if __name__ == "__main__" block lets you run the ETL directly with python etl.py. It also allows other scripts to import run_etl() without triggering the pipeline automatically.
  • Save etl.py.

Before you run the full ETL, think about what you expect to happen. You already saw how many events the extract step pulled. Do you think all of them will load successfully, or will some fail?

  • Run the full ETL pipeline by running:
python etl.py

You should see output like Loaded 347 events into security_logs.db. with your own event count.

  • Verify the data landed in DuckDB by running:
python -c "import duckdb; con = duckdb.connect('security_logs.db'); print(con.sql('SELECT COUNT(*) FROM events').fetchone())"

You should see a tuple like (347,) with a count greater than zero. That confirms your ETL pipeline extracted, transformed, and loaded real Windows Event Log data into a queryable DuckDB database.

Seeing an error when running the ETL?

  • If you see IndentationError, check that the con.executemany() block is indented inside load_events() at the same level as the if not events: block above it.
  • If you see FileNotFoundError, confirm that C:\logs\windows_events.log exists and that NXLog has written events to it.
  • If you see No new events to load., the extract step found no valid events. Delete etl_state.txt to reset the byte offset and run again.

Help me debug my ETL pipeline that loads NXLog events into DuckDB

Why DuckDB instead of SQLite?

DuckDB is purpose-built for analytical queries. It uses columnar storage and vectorized execution, which makes aggregating thousands of events fast. SQLite would work for storing data, but it is optimized for transactional workloads (single-row inserts and lookups).

DuckDB is increasingly common in data engineering job listings. Building with it here gives you a relevant skill and a portfolio project that speaks the industry's language.

✔️ Awesome, I've got everything!

Great. Double check you have saved etl.py and that your python etl.py output showed a non-zero event count.

ⓧ I'd like to double check the full code

Since etl.py is a longer file, scroll up and confirm each section is present in this order:

  • Imports (json, os, datetime, duckdb) and file path constants (LOG_FILE, DB_FILE, STATE_FILE).
  • SEVERITY_MAP dictionary with 7 entries (CRITICAL through AUDIT_FAILURE).
  • get_last_position() and save_position() for byte offset tracking.
  • classify_severity() and parse_event_time() transform helpers.
  • extract_events() function with JSON parsing and field extraction.
  • load_events() function with table creation and executemany() insertion.
  • run_etl() orchestration function and if __name__ == "__main__" entry point.

Your ETL pipeline is working. Real Windows Event Logs are now sitting in a queryable DuckDB database. Next up, you will automate this pipeline so it runs every time NXLog writes new data, without you lifting a finger.

Add Automated File Watching with watchdog

Right now, every time NXLog writes new events to your log file, you have to manually run python etl.py to process them.

In production, data engineers build systems that react to new data automatically. The watchdog library monitors your log directory and triggers the ETL whenever NXLog appends new events, turning your manual script into an automated pipeline.

In this step, get ready to:
  • Create a file watcher that detects new log data using watchdog.
  • Wire the watcher to your ETL pipeline with error handling and cooldown logic.
  • Test the full automated pipeline end-to-end.
Create the log file event handler

watchdog works on an event-driven model. You define a handler class that reacts to file system changes, and watchdog calls your handler whenever something happens in the watched directory.

  • Create a new file called watcher.py in your project directory (the same folder where etl.py lives) and add the following code:
import time

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

from etl import run_etl

# Directory where NXLog writes log files
WATCH_DIR = r"C:\logs"

class LogFileHandler(FileSystemEventHandler):
    def __init__(self):
        self.last_run = 0  # Timestamp of last ETL run
        self.cooldown = 5  # Seconds to wait between runs

    def on_modified(self, event):
        if event.is_directory:
            return
        if not event.src_path.endswith(".log"):
            return
        # Debounce rapid file-change events
        now = time.time()
        if now - self.last_run < self.cooldown:
            return
        self.last_run = now
        print(f"New log data detected, running ETL...")
        # Run ETL but catch errors so the watcher keeps running
        try:
            run_etl()
        except Exception as e:
            print(f"ETL error: {e}")

What does this code do?

LogFileHandler extends watchdog's FileSystemEventHandler. When any file in the watched directory changes, watchdog calls on_modified().

  • The handler ignores directory changes and any files that are not .log files.
  • A 5-second cooldown (self.cooldown) debounces rapid file changes. NXLog can trigger many write events in quick succession, so the cooldown prevents the ETL from running dozens of times per second.
  • The try/except around run_etl() catches errors and prints them instead of crashing the watcher. The observer keeps running even if a single ETL run fails.

What about duplicate records?

The ETL you built in Step 3 already handles deduplication. It tracks the last byte offset in etl_state.txt and only processes new lines added since the last run. The watcher can safely call run_etl() repeatedly without creating duplicate records.

  • Save watcher.py.
Wire the observer and run the watcher

The handler knows what to do when a file changes. Now you need an Observer that actually watches the directory and dispatches events to your handler.

  • In watcher.py, add the main() function below the LogFileHandler class:
def main():
    # Set up the file watcher
    handler = LogFileHandler()
    observer = Observer()
    observer.schedule(handler, WATCH_DIR, recursive=False)
    observer.start()
    print(f"Watching {WATCH_DIR} for new log data. Press Ctrl+C to stop.")
    # Keep running until Ctrl+C
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()


if __name__ == "__main__":
    main()

What does this code do?

  • observer.schedule(handler, WATCH_DIR, recursive=False) tells the observer to monitor C:\logs for changes and send events to your LogFileHandler. The recursive=False flag means it only watches that directory, not subfolders.
  • The while True: time.sleep(1) loop keeps the script alive. Without it, the script would exit immediately after starting the observer.
  • When you press Ctrl+C, a KeyboardInterrupt triggers observer.stop() and observer.join() for a clean shutdown.

✔️ Awesome, I've got everything!

Great. Double check you've saved watcher.py.

ⓧ I'd like to double check the full code

Scroll up and confirm these sections are present in watcher.py, in this order:

  • The imports (time, FileSystemEventHandler, Observer, run_etl) and the WATCH_DIR constant.
  • The LogFileHandler class with __init__ and on_modified methods.
  • The main() function and the if __name__ == "__main__" block at the bottom.
  • Save watcher.py.
  • In your terminal, make sure you are in your project directory and your virtual environment is active. Start the watcher by running:
python watcher.py

You should see: Watching C:\logs for new log data. Press Ctrl+C to stop.

The watcher is now running as a long-lived process. It will keep your terminal occupied until you press Ctrl+C.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is active before running the command. You should see the environment name in parentheses at the start of your terminal prompt.
  • Confirm that watcher.py is in the same folder as etl.py. The from etl import run_etl line needs both files in the same directory.

Help me debug my watcher.py ModuleNotFoundError

Test the pipeline with new events

The watcher is running, but it has not detected any file changes yet. To test the full automated pipeline, you need to generate new events so NXLog writes to the log file.

Before you trigger new events, what do you think will happen in the watcher terminal when NXLog appends new lines to the log file?

  • Leave the watcher running in its terminal. Open a second terminal as Administrator (right-click your terminal app and select Run as administrator).
  • Restart the NXLog service to generate a burst of new events by running these commands:
net stop nxlog
net start nxlog
  • Switch back to the terminal where watcher.py is running.

Within a few seconds, the watcher detects the file change and triggers the ETL. You should see output like:

New log data detected, running ETL...
Loaded 150 events into security_logs.db.

What just happened?

The full automated pipeline ran end-to-end. NXLog wrote new events to C:\logs\windows_events.log. watchdog detected the file change and called on_modified(). The handler's cooldown let the burst finish, then triggered run_etl(). The ETL read only the new lines (thanks to the byte-offset tracker) and loaded them into DuckDB.

Not seeing any output in the watcher terminal?

  • Wait at least 10 seconds after restarting NXLog. The service needs time to start collecting and writing events.
  • Confirm that C:\logs\windows_events.log exists and is growing. Open it in VS Code and check that new JSON lines are appearing.
  • Make sure the WATCH_DIR in watcher.py matches r"C:\logs" exactly. A typo in the path means watchdog is monitoring a different (or nonexistent) folder.

Help me debug why my watcher is not detecting file changes

  • Stop the watcher by pressing Ctrl+C in its terminal.

Your pipeline is now fully automated. NXLog collects events, watchdog detects the new data, and the ETL processes it into DuckDB without you lifting a finger. Next up, you will build a Streamlit dashboard to visualize all that security event data.

Build the Interactive Streamlit Security Dashboard

Your ETL pipeline now runs hands-free. The watcher detects new log data from NXLog CE and triggers the ETL automatically, loading clean records into DuckDB without you lifting a finger.

There is one problem. The pipeline is invisible to anyone who is not a developer. A stakeholder, a SOC analyst, or a hiring manager cannot open a DuckDB file and draw conclusions from it. In this step, you will build an interactive Streamlit dashboard with Plotly charts that turns your pipeline's output into something anyone can see and explore.

In this step, get ready to:
  • Connect DuckDB to a Streamlit app with sidebar filters for severity and event channel.
  • Add Plotly visualizations showing event timelines, severity breakdowns, and top event sources.
  • Build a data explorer table for drilling into individual events.
Create the dashboard layout with sidebar filters and metric cards

Streamlit apps are plain Python scripts. You build the UI by calling Streamlit functions in order, and Streamlit renders each call as a widget on the page.

  • Create a new file called dashboard.py in your project directory and paste the following code:
import duckdb
import plotly.express as px
import streamlit as st

# Path to the DuckDB database populated by the ETL pipeline
DB_FILE = "security_logs.db"

# Cache the connection so Streamlit reuses it across reruns
@st.cache_resource
def get_connection():
    return duckdb.connect(DB_FILE, read_only=True)

def main():
    # Set up a wide-layout page for the dashboard
    st.set_page_config(
        page_title="Security Log Dashboard",
        layout="wide",
    )
    st.title("Security Log ETL Dashboard")
    st.caption("Real Windows Event Logs collected by NXLog CE")

if __name__ == "__main__":
    main()

What does this code do?

get_connection() opens your security_logs.db file in read-only mode so the dashboard can query data without accidentally modifying it. The @st.cache_resource decorator tells Streamlit to create the connection once and reuse it every time the page reruns.

st.set_page_config() sets the browser tab title and switches to a wide layout so your charts have room to breathe.

  • Save dashboard.py.
  • Make sure your Python virtual environment is activated, then launch the dashboard by running:
streamlit run dashboard.py

Streamlit starts a local web server and opens your browser to http://localhost:8501. You should see the title "Security Log ETL Dashboard" and a caption below it.

Streamlit auto-reruns on save

Streamlit watches your Python file for changes. Every time you save dashboard.py in VS Code, Streamlit automatically reruns the script and updates the browser. Keep the terminal running and switch back to VS Code for the next chunks.

Seeing a database error or blank page?

  • If you see a FileNotFoundError or database error, make sure you are running the command from your project directory (the one containing security_logs.db).
  • If streamlit is not recognized, confirm your virtual environment is activated.

Why can't Streamlit find my database?

Now you need sidebar filters so users can narrow down the events they see. The sidebar will query the database for all distinct severity levels and event channels, then present them as multiselect dropdowns.

  • In dashboard.py, find this line inside main():
    st.caption("Real Windows Event Logs collected by NXLog CE")
  • Add the following code directly below that line, above the blank line before the if __name__ guard:
    con = get_connection()

    # Query the database for the unique filter values
    severities = con.sql(
        "SELECT DISTINCT severity FROM events ORDER BY severity"
    ).fetchdf()["severity"].tolist()
    channels = con.sql(
        "SELECT DISTINCT channel FROM events ORDER BY channel"
    ).fetchdf()["channel"].tolist()

    # Build the sidebar with filter controls
    with st.sidebar:
        st.header("Filters")
        selected_severities = st.multiselect(
            "Severity", severities, default=severities
        )
        selected_channels = st.multiselect(
            "Channel", channels, default=channels
        )
        if st.button("Refresh Data"):
            st.cache_resource.clear()
            st.rerun()

    # Build a SQL WHERE clause from the selected filters
    sev_list = ", ".join(f"'{s}'" for s in selected_severities)
    chan_list = ", ".join(f"'{c}'" for c in selected_channels)
    where = f"""
        WHERE severity IN ({sev_list})
        AND channel IN ({chan_list})
    """

What does this code do?

  • .fetchdf()["severity"].tolist() runs the SQL query, converts the result to a Pandas DataFrame, and extracts the column as a plain Python list. This gives Streamlit's st.multiselect() the options to display.
  • The Refresh Data button clears the cached database connection and forces Streamlit to rerun the entire script. This picks up any new events the watcher has loaded since you opened the dashboard.
  • The where variable builds a reusable SQL WHERE clause from the selected filters. Every query in the dashboard appends this clause so all charts and tables stay in sync.
  • Save dashboard.py.
  • Switch to your browser. You should see a Filters panel in the left sidebar with two multiselect dropdowns (Severity and Channel) and a Refresh Data button.

Sidebar not appearing?

  • Check that the new code is indented with 4 spaces so it sits inside the main() function. Code outside main() will not run.
  • If Streamlit shows an error about the events table not existing, re-run python etl.py to populate security_logs.db first.

Why is my Streamlit sidebar empty or throwing errors?

The sidebar filters work, but the main page is still empty. Time to add summary metrics so users can see the big picture at a glance.

  • In dashboard.py, find the closing triple-quote of the where variable:
    where = f"""
        WHERE severity IN ({sev_list})
        AND channel IN ({chan_list})
    """
  • Add the following code directly below that block to query aggregate metrics and display them as cards:
    # Query aggregate metrics for the header cards
    total = con.sql(
        f"SELECT COUNT(*) FROM events {where}"
    ).fetchone()[0]
    sources = con.sql(
        f"SELECT COUNT(DISTINCT source_name) FROM events {where}"
    ).fetchone()[0]
    top_sev = con.sql(
        f"""SELECT severity, COUNT(*) as cnt FROM events {where}
        GROUP BY severity ORDER BY cnt DESC LIMIT 1"""
    ).fetchone()

    # Display three metric cards in a row
    col1, col2, col3 = st.columns(3)
    col1.metric("Total Events", f"{total:,}")
    col2.metric("Unique Sources", f"{sources:,}")
    col3.metric(
        "Most Common Severity", top_sev[0] if top_sev else "N/A"
    )

What does this code do?

  • Each con.sql() call runs a DuckDB query that respects the current sidebar filters through the where clause. .fetchone()[0] grabs the single numeric result from each query.
  • st.columns(3) creates three equal-width columns. Each column displays a metric card with a label and a value, giving an instant summary of event volume, source diversity, and the dominant severity level.
  • Save dashboard.py and switch to your browser.
  • You should see three metric cards across the top of the page: Total Events showing a number in the thousands, Unique Sources, and Most Common Severity.

Metrics showing zero or N/A?

  • Check the sidebar filters. If you deselected all severity levels or channels, the WHERE clause filters out every row. Re-select all options to restore the full dataset.
  • Verify that security_logs.db has data by running python etl.py again in a separate terminal.

Why are my dashboard metrics showing zero?

Add Plotly visualizations for event trends and severity

Metric cards give the big picture, but charts reveal patterns over time. You will add three Plotly charts: an event timeline, a severity breakdown, and a top sources ranking.

  • In dashboard.py, find the last metric card line:
    col3.metric(
        "Most Common Severity", top_sev[0] if top_sev else "N/A"
    )
  • Add the event timeline chart directly below that line:
    # Create two side-by-side columns for the charts
    chart_col1, chart_col2 = st.columns(2)

    with chart_col1:
        st.subheader("Events Over Time")
        # Count events per hour using DuckDB date_trunc
        timeline_df = con.sql(f"""
            SELECT date_trunc('hour', event_time) AS hour,
                   COUNT(*) AS count
            FROM events {where}
            GROUP BY hour ORDER BY hour
        """).fetchdf()
        if not timeline_df.empty:
            fig = px.line(
                timeline_df, x="hour", y="count",
                labels={"hour": "Time", "count": "Events"},
            )
            st.plotly_chart(fig, use_container_width=True)
        else:
            st.info("No events for the selected filters.")

What does this code do?

  • date_trunc('hour', event_time) is a DuckDB function that rounds each timestamp down to the hour. Grouping by this truncated value counts how many events occurred each hour, which is the data behind the line chart.
  • px.line() creates a Plotly line chart. The if not timeline_df.empty guard prevents Plotly from crashing on an empty result set when all filters are deselected.
  • Save dashboard.py and check your browser.
  • You should see a line chart titled "Events Over Time" on the left half of the page, with a time axis and event counts. The right half is still empty.

Chart not appearing?

  • Make sure the new code is inside main() with 4-space indentation. The with chart_col1: block must be at the same indent level as the col1, col2, col3 line above it.
  • If you see "No events for the selected filters" instead of a chart, check that all sidebar filters are selected.

Why is my Plotly chart not rendering in Streamlit?

The right column is waiting for a severity breakdown. A pie chart is a natural fit here because severity levels are a small, finite set of categories.

  • In dashboard.py, find this line at the end of the timeline chart block:
            st.info("No events for the selected filters.")
  • Add the severity pie chart directly below that line:
    with chart_col2:
        st.subheader("Severity Distribution")
        # Count events by severity level
        sev_df = con.sql(f"""
            SELECT severity, COUNT(*) AS count
            FROM events {where} GROUP BY severity
        """).fetchdf()
        if not sev_df.empty:
            # Color each severity slice with a distinct color
            fig = px.pie(
                sev_df, names="severity", values="count",
                color="severity",
                color_discrete_map={
                    "Critical": "#d62728",
                    "High": "#ff7f0e",
                    "Medium": "#ffdd57",
                    "Low": "#2ca02c",
                    "Info": "#1f77b4",
                },
            )
            st.plotly_chart(fig, use_container_width=True)
        else:
            st.info("No events for the selected filters.")

What does the color map do?

color_discrete_map assigns a fixed color to each severity level. Red for Critical, orange for High, yellow for Medium, green for Low, and blue for Info. This ensures the colors stay consistent no matter how the data changes.

  • Save dashboard.py and check your browser.
  • You should now see the severity pie chart on the right, next to the timeline chart. The pie slices show the proportion of each severity level in your log data.

Pie chart overlapping the timeline?

  • The with chart_col2: block must be at the same indentation level as with chart_col1: (4 spaces inside main()). If it is nested inside chart_col1, both charts will stack in one column.

Why are my Streamlit columns not side-by-side?

The last chart shows which event sources generate the most activity. A horizontal bar chart makes long source names easy to read.

  • In dashboard.py, find the last line of the chart_col2 block:
            st.info("No events for the selected filters.")
  • Add the top event sources bar chart directly below that line:
    st.subheader("Top 10 Event Sources")
    # Find the most active event sources
    sources_df = con.sql(f"""
        SELECT source_name, COUNT(*) AS count
        FROM events {where}
        GROUP BY source_name ORDER BY count DESC LIMIT 10
    """).fetchdf()
    if not sources_df.empty:
        fig = px.bar(
            sources_df, x="count", y="source_name",
            orientation="h",
            labels={"source_name": "Source", "count": "Events"},
        )
        fig.update_layout(yaxis={"categoryorder": "total ascending"})
        st.plotly_chart(fig, use_container_width=True)

What does this code do?

  • orientation="h" flips the bar chart to horizontal, so the source names appear on the y-axis where they have room to display.
  • categoryorder: "total ascending" sorts the bars so the source with the most events appears at the top, making it easy to spot the noisiest sources.
  • Save dashboard.py and check your browser.
  • A horizontal bar chart titled "Top 10 Event Sources" should appear below the two side-by-side charts, spanning the full width of the page.

Bar chart rendering inside one of the columns?

  • The st.subheader("Top 10 Event Sources") line must be at 4-space indentation (inside main() but outside both chart_col1 and chart_col2). If it is indented 8 spaces, it is still inside a column block.

Why is my bar chart stuck inside a column instead of full width?

Add the event explorer table

Charts show trends, but sometimes you need to inspect individual events. An explorer table at the bottom gives users a way to drill into the raw data.

  • In dashboard.py, find this line at the end of the bar chart block:
        st.plotly_chart(fig, use_container_width=True)
  • Add the event explorer table directly below that line:
    # Display a table of recent events for drill-down
    st.subheader("Event Explorer")
    page_size = 25
    explorer_df = con.sql(f"""
        SELECT event_time, event_id, severity, source_name,
               channel, message
        FROM events {where}
        ORDER BY event_time DESC LIMIT {page_size}
    """).fetchdf()
    st.dataframe(explorer_df, use_container_width=True)

What does this code do?

The query selects the 25 most recent events that match the current filters, ordered newest-first. st.dataframe() renders them as a sortable, scrollable table. You can click any column header to re-sort the data right in the browser.

  • Save dashboard.py.
  • Scroll to the bottom of your dashboard in the browser. You should see an "Event Explorer" table with columns for event time, event ID, severity, source name, channel, and message.

Event explorer table not showing?

  • Make sure st.subheader("Event Explorer") is at 4-space indentation inside main(), not nested inside the bar chart's if block.
  • Confirm that the if __name__ == "__main__": guard is still the very last block in the file, below all the code you just added.

Why is my Streamlit dataframe table not rendering?

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboard.py and the dashboard is displaying all sections in your browser.

ⓧ I'd like to double check the full code

Scroll through your dashboard.py and confirm these sections are present in order:

  • Imports (duckdb, plotly.express, streamlit) and the DB_FILE constant.
  • get_connection() function with the @st.cache_resource decorator.
  • main() function starting with st.set_page_config(), title, and caption.
  • Distinct value queries for severities and channels, followed by the sidebar with multiselects and the Refresh Data button.
  • The where clause builder, then three metric card queries and st.columns(3) to display them.
  • Two chart columns: the event timeline (px.line) in chart_col1 and the severity pie chart (px.pie) in chart_col2.
  • The full-width top 10 event sources bar chart (px.bar).
  • The Event Explorer st.dataframe() section.
  • The if __name__ == "__main__": guard at the very end.

Before you confirm everything works end-to-end, take a moment to predict. What do you expect to see when you try changing the sidebar filters right now?

  • In the sidebar, deselect one or two severity levels and watch what happens to the charts and metrics.

Every chart, metric card, and the explorer table update instantly to reflect only the selected filters. This is because every query in the dashboard uses the same where clause built from the sidebar selections.

Why read-only mode?

The dashboard opens DuckDB with read_only=True so it can never accidentally write to or corrupt security_logs.db. The ETL pipeline owns the writes. The dashboard only reads. This separation means you can safely run both at the same time.

Your full end-to-end pipeline is now working. NXLog collects Windows Event Logs, the watcher triggers the ETL, DuckDB stores the data, and Streamlit visualizes it. Next up, you have an optional Secret Mission to take your analytics even further.

Secret mission

Add Anomaly Frequency Detection

Your dashboard shows what happened, but can it spot when something unusual is happening? In this mission, you will write a DuckDB window function that computes rolling averages and standard deviations to flag event frequency spikes. You will surface those anomalies in a dedicated tab and color-code them on your timeline chart.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local Windows machine using open-source tools. There are no cloud resources or ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • NXLog CE Windows service (continuously collecting Windows Event Logs to C:\logs\windows_events.log).
  • C:\logs\ directory containing log output files.
  • Project folder on your Desktop containing etl.py, watcher.py, dashboard.py, security_logs.db, etl_state.txt, and the Python virtual environment.

Keep everything running

No action needed. Choose this if you are still actively building or want the pipeline to keep processing new logs.

  • NXLog CE continues collecting Windows Event Logs in the background as a Windows service.
  • If the watcher script is running, it continues processing new log data into DuckDB automatically.
  • The security_logs.db database file grows over time, but Windows Event Logs are lightweight so it stays small.

Pause - I'll come back to this later

Shut down running processes to free up resources, but keep all files and data on disk so you can pick up where you left off.

  • Stop the watcher script by pressing Ctrl+C in its terminal window.
  • Stop the NXLog CE service by opening an Administrator terminal and running the command below.
net stop nxlog
  • NXLog will start again automatically when you reboot your machine. Your log files, database, and project code all stay on disk.
  • When you are ready to resume, run net start nxlog in an Administrator terminal to restart log collection. Then run python watcher.py to resume automated ETL processing.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below to fully clean up.

  • Stop the watcher script by pressing Ctrl+C in its terminal window (if it is still running).
  • Stop and disable the NXLog CE service by running these commands in an Administrator terminal:
net stop nxlog
sc config nxlog start= disabled
  • Delete the log output directory by running this command in an Administrator terminal:
rmdir /s /q C:\logs
  • Delete your project folder from the Desktop. Open File Explorer, navigate to your Desktop, right-click the project folder that contains etl.py and dashboard.py, and select Delete.
  • Uninstall NXLog CE from your machine.
  • Press the Windows key and type Add or remove programs. Press Enter.
  • Search for NXLog in the apps list.
  • Click the three-dot menu next to NXLog-CE and select Uninstall.
  • Follow the prompts to complete the uninstall.

Nice Work!

Nice Work!

You just built a complete, end-to-end security log ETL pipeline that collects real Windows Event Logs from your own machine, transforms them into structured analytical data, and visualizes the results through an interactive dashboard.

You've learned how to:

  • Configured NXLog CE to collect live Windows Event Logs from Security, System, and Application channels and output them as structured JSON.
  • Built a Python ETL pipeline that parses nested JSON log data, classifies severity levels, normalizes timestamps, and loads clean records into a DuckDB analytical database.
  • Automated the entire pipeline with watchdog file monitoring and built an interactive Streamlit dashboard with Plotly charts to explore security event trends, severity distributions, and top event sources.
  • Secret Mission: Added anomaly frequency detection using DuckDB window functions and surfaced spike alerts with red markers on the timeline chart.

Ready to quiz yourself?