From SIEM to Lakehouse: Detection-as-Code on Databricks

Table of Contents

Every SIEM I’ve worked with treats detection rules the same way: you click through a GUI, write some vendor-specific query language, hit save, and pray it doesn’t fire 10,000 false positives on Monday morning. No version history. No tests. No peer review. No rollback.

I’ve spent years migrating security operations from traditional SIEMs to data lake architectures. The biggest unlock wasn’t the storage or the compute. It was the moment we started treating detection rules like software.

This post walks through a Detection-as-Code framework built entirely on Databricks: rules in YAML, tests against real data in Delta Lake, CI/CD via Databricks Workflows, and a Databricks App as the management UI. Everything runs in the Lakehouse. No external tools required.

Why SIEM-style detection doesn’t scale
#

Traditional SIEM detection has three problems that get worse with scale:

1. Tribal knowledge. Rules live in the SIEM’s internal database. The only documentation is what someone typed into a description field three years ago. When that person leaves, the rule becomes an oracle nobody dares touch.

2. No testing. You find out a rule is broken when it either floods the SOC with noise or misses an actual incident. There’s no staging environment, no unit tests, no way to validate a rule against historical data before deploying it.

3. Vendor lock-in. Every SIEM has its own query language. SPL, KQL, AQL, YARA-L. Migrating 500 rules from Splunk to Sentinel isn’t a weekend project. It’s a six-month program with a dedicated team.

The Lakehouse changes all three. Your data is in Delta tables. Your compute is SQL or Python. Your deployment is code.

The framework
#

flowchart LR
    subgraph Dev["Development"]
        R["Rule YAML\n+ Python"] --> Git["Git Repository"]
    end

    subgraph CICD["CI/CD"]
        Git --> PR["Pull Request"]
        PR --> Test["Unit Tests\n(historical data)"]
        Test --> Review["Peer Review"]
        Review --> Merge["Merge to main"]
    end

    subgraph Databricks["Databricks"]
        Merge --> WF["Databricks Workflow\n(deploy)"]
        WF --> Cat["Unity Catalog\n(rule store)"]
        Cat --> RT["Scheduled Jobs\n(rule execution)"]
        RT --> Delta["Delta Tables\n(alerts)"]
    end

    subgraph UI["Management"]
        App["Databricks App"] --> Cat
        App --> Delta
    end

    style Test fill:#6366f1,color:#fff
    style App fill:#6366f1,color:#fff

Four components:

Rule format — YAML with embedded SQL/Python, metadata, and MITRE mapping
Test harness — run rules against historical Delta Lake data, validate expected matches
Deployment pipeline — Databricks Workflow triggered on merge to main
Management App — Databricks App for browsing rules, coverage, and test results

1. Rule format
#

Every rule is a single YAML file. The format is deliberately simple — anyone who can write SQL can write a detection rule.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45


# rules/brute_force_ssh.yaml
id: DET-001
name: SSH Brute Force
version: 2
status: active
severity: high

mitre:
  tactic: Credential Access
  technique: T1110.001
  name: Brute Force - Password Guessing

data_sources:
  - auth_logs
  - network_flows

sql: |
  SELECT
    src_ip,
    dst_ip,
    dst_port,
    COUNT(*) AS attempt_count,
    COUNT(DISTINCT user_name) AS distinct_users,
    MIN(event_time) AS first_seen,
    MAX(event_time) AS last_seen
  FROM {catalog}.{schema}.auth_logs
  WHERE event_time >= CURRENT_TIMESTAMP - INTERVAL {lookback}
    AND action = 'login_failed'
    AND dst_port = 22
  GROUP BY src_ip, dst_ip, dst_port
  HAVING attempt_count >= {threshold}

parameters:
  lookback: "1 HOUR"
  threshold: 20

false_positive_notes: |
  - Service accounts with automated SSH key rotation
  - Vulnerability scanners on scheduled scans
  - Jump hosts with high legitimate traffic

tags:
  - linux
  - ssh
  - network

Why YAML + SQL
#

YAML because it’s human-readable, diffable, and parseable. Every change shows up cleanly in a PR.
SQL because it’s the lingua franca. Your SOC analyst doesn’t need to learn a new DSL. If they can query the data in a notebook, they can write a rule.
Parameterized queries ({lookback}, {threshold}) so the same rule can run with different settings per environment without code changes.

The {catalog}.{schema} placeholders are resolved at deployment time from the environment configuration, so the same rule works across dev/staging/prod Unity Catalog schemas.

MITRE ATT&CK mapping
#

Every rule maps to at least one MITRE technique. This isn’t documentation for the sake of documentation — it’s what makes the coverage dashboard possible. More on that in section 4.

2. Testing rules against historical data
#

This is where Detection-as-Code earns its name. Every rule has tests. Tests run against real data in Delta Lake.

Test format
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31


# tests/brute_force_ssh_test.yaml
rule: DET-001
description: "Validate SSH brute force detection against known incidents"

test_cases:
  - name: "Known brute force from 2025-12-15 incident"
    time_range:
      start: "2025-12-15T02:00:00Z"
      end: "2025-12-15T04:00:00Z"
    expect:
      min_results: 1
      must_contain:
        src_ip: "10.45.2.118"
      severity_override: null

  - name: "Clean period - no false positives expected"
    time_range:
      start: "2025-11-01T10:00:00Z"
      end: "2025-11-01T12:00:00Z"
    expect:
      max_results: 0

  - name: "Scanner exclusion - should not trigger"
    time_range:
      start: "2025-12-20T00:00:00Z"
      end: "2025-12-20T06:00:00Z"
    parameters:
      threshold: 20
    expect:
      must_not_contain:
        src_ip: "10.100.0.5" # vulnerability scanner

Three types of assertions:

True positive — the rule must fire on a known incident
True negative — the rule must stay quiet during a clean period
Exclusion — known benign sources must not trigger

Test runner
#

The test runner is a Python module that loads the rule YAML, resolves parameters, executes the SQL against a Delta table with a time filter, and validates assertions.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77


# detection_framework/test_runner.py
import yaml
from pyspark.sql import SparkSession
from datetime import datetime

class DetectionTestRunner:
    def __init__(self, spark: SparkSession, catalog: str, schema: str):
        self.spark = spark
        self.catalog = catalog
        self.schema = schema

    def load_rule(self, rule_path: str) -> dict:
        with open(rule_path) as f:
            return yaml.safe_load(f)

    def execute_rule(self, rule: dict, params: dict = None) -> list:
        resolved_params = {**rule["parameters"], **(params or {})}
        resolved_params["catalog"] = self.catalog
        resolved_params["schema"] = self.schema

        sql = rule["sql"].format(**resolved_params)
        return self.spark.sql(sql).collect()

    def run_test_case(self, rule: dict, test_case: dict) -> dict:
        # Override time range via parameter injection
        params = test_case.get("parameters", {})
        time_range = test_case["time_range"]

        # Wrap the rule SQL with a time-bounded CTE
        original_sql = rule["sql"]
        bounded_sql = f"""
        WITH time_bounded AS (
            {original_sql}
        )
        SELECT * FROM time_bounded
        WHERE first_seen >= '{time_range["start"]}'
          AND last_seen <= '{time_range["end"]}'
        """
        rule_copy = {**rule, "sql": bounded_sql}
        results = self.execute_rule(rule_copy, params)

        # Validate assertions
        expect = test_case["expect"]
        failures = []

        if "min_results" in expect and len(results) < expect["min_results"]:
            failures.append(
                f"Expected >= {expect['min_results']} results, got {len(results)}"
            )

        if "max_results" in expect and len(results) > expect["max_results"]:
            failures.append(
                f"Expected <= {expect['max_results']} results, got {len(results)}"
            )

        if "must_contain" in expect:
            for key, value in expect["must_contain"].items():
                found = any(
                    str(getattr(r, key, None)) == str(value) for r in results
                )
                if not found:
                    failures.append(f"Expected {key}={value} in results")

        if "must_not_contain" in expect:
            for key, value in expect["must_not_contain"].items():
                found = any(
                    str(getattr(r, key, None)) == str(value) for r in results
                )
                if found:
                    failures.append(f"Unexpected {key}={value} found in results")

        return {
            "test_name": test_case["name"],
            "passed": len(failures) == 0,
            "result_count": len(results),
            "failures": failures,
        }

Running tests in CI
#

Tests execute as a Databricks Workflow step, triggered on every PR. The job runs in a dedicated detection_test schema with read-only access to production data.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


# databricks_workflows/test_detection_rules.yaml
name: "detection-rules-test"
tasks:
  - task_key: "run_tests"
    notebook_task:
      notebook_path: "/Repos/security/detection-rules/notebooks/run_tests"
      base_parameters:
        catalog: "security"
        schema: "detection_test"
        rules_path: "/Repos/security/detection-rules/rules"
        tests_path: "/Repos/security/detection-rules/tests"
    new_cluster:
      spark_version: "15.4.x-scala2.12"
      num_workers: 2
      node_type_id: "Standard_D4ds_v5"

The test notebook iterates over all test files, runs each case, and fails the workflow if any assertion breaks:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60


# notebooks/run_tests.py
import os
import yaml
import sys

dbutils.widgets.text("catalog", "security")
dbutils.widgets.text("schema", "detection_test")
dbutils.widgets.text("rules_path", "")
dbutils.widgets.text("tests_path", "")

catalog = dbutils.widgets.get("catalog")
schema = dbutils.widgets.get("schema")
rules_path = dbutils.widgets.get("rules_path")
tests_path = dbutils.widgets.get("tests_path")

runner = DetectionTestRunner(spark, catalog, schema)

all_results = []
failed = False

for test_file in os.listdir(tests_path):
    if not test_file.endswith("_test.yaml"):
        continue

    with open(os.path.join(tests_path, test_file)) as f:
        test_spec = yaml.safe_load(f)

    rule_id = test_spec["rule"]

    # Find the corresponding rule file
    rule = None
    for rule_file in os.listdir(rules_path):
        with open(os.path.join(rules_path, rule_file)) as f:
            candidate = yaml.safe_load(f)
            if candidate["id"] == rule_id:
                rule = candidate
                break

    if rule is None:
        print(f"SKIP: Rule {rule_id} not found for test {test_file}")
        continue

    for test_case in test_spec["test_cases"]:
        result = runner.run_test_case(rule, test_case)
        all_results.append(result)
        status = "PASS" if result["passed"] else "FAIL"
        print(f"  [{status}] {result['test_name']}")
        if not result["passed"]:
            failed = True
            for f in result["failures"]:
                print(f"         -> {f}")

# Write results to Delta for the App dashboard
results_df = spark.createDataFrame(all_results)
results_df.write.mode("overwrite").saveAsTable(
    f"{catalog}.{schema}.test_results"
)

if failed:
    raise Exception("Detection rule tests failed. See output above.")

3. Deployment pipeline
#

When tests pass and the PR is merged, a deployment workflow picks up the changes and registers the rules in Unity Catalog.

Rule catalog table
#

Rules are stored as a Delta table in Unity Catalog. This is the single source of truth for what’s active in production.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


CREATE TABLE IF NOT EXISTS security.detection.rules (
    rule_id STRING,
    name STRING,
    version INT,
    status STRING,
    severity STRING,
    mitre_tactic STRING,
    mitre_technique STRING,
    mitre_name STRING,
    data_sources ARRAY<STRING>,
    sql_template STRING,
    parameters MAP<STRING, STRING>,
    false_positive_notes STRING,
    tags ARRAY<STRING>,
    deployed_at TIMESTAMP,
    deployed_by STRING,
    git_commit STRING
);

Deploy notebook
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36


# notebooks/deploy_rules.py
import yaml
import os
from datetime import datetime

rules_path = dbutils.widgets.get("rules_path")
catalog = dbutils.widgets.get("catalog")
git_commit = dbutils.widgets.get("git_commit")

for rule_file in os.listdir(rules_path):
    if not rule_file.endswith(".yaml"):
        continue

    with open(os.path.join(rules_path, rule_file)) as f:
        rule = yaml.safe_load(f)

    spark.sql(f"""
        MERGE INTO {catalog}.detection.rules AS target
        USING (SELECT
            '{rule["id"]}' AS rule_id,
            '{rule["name"]}' AS name,
            {rule["version"]} AS version,
            '{rule["status"]}' AS status,
            '{rule["severity"]}' AS severity,
            '{rule["mitre"]["tactic"]}' AS mitre_tactic,
            '{rule["mitre"]["technique"]}' AS mitre_technique,
            '{rule["mitre"]["name"]}' AS mitre_name,
            '{rule["sql"]}' AS sql_template,
            '{git_commit}' AS git_commit
        ) AS source
        ON target.rule_id = source.rule_id
        WHEN MATCHED AND target.version < source.version THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

    print(f"Deployed {rule['id']} v{rule['version']} ({rule['status']})")

Scheduled execution
#

Each active rule runs on a schedule via Databricks Workflows. The execution notebook reads rules from the catalog table, runs the SQL, and writes alerts to a Delta table.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38


# notebooks/execute_rules.py
catalog = dbutils.widgets.get("catalog")
schema = dbutils.widgets.get("schema")

active_rules = spark.sql(f"""
    SELECT * FROM {catalog}.detection.rules
    WHERE status = 'active'
""").collect()

for rule in active_rules:
    try:
        params = rule.parameters or {}
        params["catalog"] = catalog
        params["schema"] = schema
        sql = rule.sql_template.format(**params)

        alerts = spark.sql(sql)
        alert_count = alerts.count()

        if alert_count > 0:
            (alerts
                .withColumn("rule_id", lit(rule.rule_id))
                .withColumn("rule_name", lit(rule.name))
                .withColumn("severity", lit(rule.severity))
                .withColumn("mitre_technique", lit(rule.mitre_technique))
                .withColumn("detected_at", current_timestamp())
                .write.mode("append")
                .saveAsTable(f"{catalog}.{schema}.alerts"))

        print(f"{rule.rule_id}: {alert_count} alerts")

    except Exception as e:
        print(f"ERROR {rule.rule_id}: {e}")
        # Write error to execution log
        spark.sql(f"""
            INSERT INTO {catalog}.{schema}.execution_log
            VALUES ('{rule.rule_id}', current_timestamp(), 'error', '{str(e)}')
        """)

4. The Databricks App
#

This is the management layer. A Databricks App that gives your SOC team visibility into rules, coverage, and test results without touching code.

App structure
#

detection-app/
  app.py          # Gradio UI
  app.yaml        # Databricks App config
  requirements.txt

app.yaml
#

1
2
3
4
5
6
7
8


command:
  - python
  - app.py
env:
  - name: DATABRICKS_CATALOG
    value: security
  - name: DATABRICKS_SCHEMA
    value: detection

The app
#

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161


# app.py
import gradio as gr
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.sql import StatementState
import json

w = WorkspaceClient()
CATALOG = "security"
SCHEMA = "detection"
WAREHOUSE_ID = "your_sql_warehouse_id"


def query(sql: str) -> list[dict]:
    """Execute SQL via Databricks SQL Statement API."""
    resp = w.statement_execution.execute_statement(
        warehouse_id=WAREHOUSE_ID,
        statement=sql,
        catalog=CATALOG,
        schema=SCHEMA,
    )
    if resp.status.state != StatementState.SUCCEEDED:
        raise Exception(f"Query failed: {resp.status.error}")

    columns = [c.name for c in resp.manifest.schema.columns]
    rows = []
    for chunk in resp.result.data_array:
        rows.append(dict(zip(columns, chunk)))
    return rows


def get_rules_overview():
    """All rules with status and last test result."""
    rows = query(f"""
        SELECT
            r.rule_id,
            r.name,
            r.severity,
            r.status,
            r.mitre_tactic,
            r.mitre_technique,
            r.version,
            r.deployed_at
        FROM {CATALOG}.{SCHEMA}.rules r
        ORDER BY r.severity DESC, r.name
    """)
    return rows


def get_mitre_coverage():
    """MITRE ATT&CK coverage matrix."""
    rows = query(f"""
        SELECT
            mitre_tactic,
            mitre_technique,
            mitre_name,
            COUNT(*) AS rule_count,
            SUM(CASE WHEN status = 'active' THEN 1 ELSE 0 END) AS active_rules
        FROM {CATALOG}.{SCHEMA}.rules
        GROUP BY mitre_tactic, mitre_technique, mitre_name
        ORDER BY mitre_tactic, mitre_technique
    """)
    return rows


def get_recent_alerts(hours: int = 24):
    """Alerts from the last N hours."""
    rows = query(f"""
        SELECT
            rule_id,
            rule_name,
            severity,
            mitre_technique,
            detected_at,
            src_ip,
            dst_ip
        FROM {CATALOG}.{SCHEMA}.alerts
        WHERE detected_at >= CURRENT_TIMESTAMP - INTERVAL {hours} HOUR
        ORDER BY detected_at DESC
        LIMIT 200
    """)
    return rows


def get_test_results():
    """Latest test run results."""
    rows = query(f"""
        SELECT
            test_name,
            passed,
            result_count,
            failures
        FROM {CATALOG}.detection_test.test_results
        ORDER BY passed ASC, test_name
    """)
    return rows


def get_coverage_stats():
    """High-level coverage statistics."""
    rows = query(f"""
        SELECT
            COUNT(DISTINCT mitre_technique) AS techniques_covered,
            COUNT(*) AS total_rules,
            SUM(CASE WHEN status = 'active' THEN 1 ELSE 0 END) AS active_rules,
            SUM(CASE WHEN severity = 'critical' THEN 1 ELSE 0 END) AS critical_rules,
            SUM(CASE WHEN severity = 'high' THEN 1 ELSE 0 END) AS high_rules
        FROM {CATALOG}.{SCHEMA}.rules
    """)
    return rows[0] if rows else {}


# --- Gradio UI ---

with gr.Blocks(
    title="Detection-as-Code",
    theme=gr.themes.Base(primary_hue="indigo"),
) as app:
    gr.Markdown("# Detection-as-Code Dashboard")
    gr.Markdown("Rule management, MITRE coverage, and alert overview.")

    with gr.Tab("Rules"):
        rules_btn = gr.Button("Refresh")
        rules_table = gr.Dataframe(
            headers=["rule_id", "name", "severity", "status",
                     "mitre_tactic", "mitre_technique", "version",
                     "deployed_at"],
            label="Detection Rules",
        )
        rules_btn.click(fn=get_rules_overview, outputs=rules_table)

    with gr.Tab("MITRE Coverage"):
        mitre_btn = gr.Button("Refresh")
        mitre_table = gr.Dataframe(
            headers=["mitre_tactic", "mitre_technique", "mitre_name",
                     "rule_count", "active_rules"],
            label="MITRE ATT&CK Coverage",
        )
        mitre_btn.click(fn=get_mitre_coverage, outputs=mitre_table)

    with gr.Tab("Alerts"):
        hours_slider = gr.Slider(1, 168, value=24, step=1,
                                 label="Lookback (hours)")
        alerts_btn = gr.Button("Refresh")
        alerts_table = gr.Dataframe(
            headers=["rule_id", "rule_name", "severity",
                     "mitre_technique", "detected_at",
                     "src_ip", "dst_ip"],
            label="Recent Alerts",
        )
        alerts_btn.click(fn=get_recent_alerts, inputs=hours_slider,
                         outputs=alerts_table)

    with gr.Tab("Test Results"):
        test_btn = gr.Button("Refresh")
        test_table = gr.Dataframe(
            headers=["test_name", "passed", "result_count", "failures"],
            label="Latest Test Run",
        )
        test_btn.click(fn=get_test_results, outputs=test_table)

app.launch()

What changes in your SOC
#

When you move from SIEM rules to this framework, the workflow shifts fundamentally:

Before (SIEM)	After (Detection-as-Code)
Rules edited in vendor GUI	Rules in YAML, versioned in Git
No review process	Every change goes through PR review
Test by deploying to prod	Unit tests against historical data
“Who changed this rule?” — nobody knows	Full Git history with author and context
Migration = rewrite everything	SQL is portable, YAML is engine-agnostic
Coverage tracking = spreadsheet	Automated MITRE mapping from rule metadata
Rule debugging = stare at logs	Replay against any time range in Delta Lake

The biggest shift is cultural. When detection is code, your security engineers start thinking like software engineers. They write tests. They review each other’s work. They refactor. They build on each other’s rules instead of duplicating them.

Getting started
#

You don’t need to migrate everything at once. Start with:

One high-value rule — pick your most critical detection (lateral movement, privilege escalation, whatever keeps you up at night). Write it in YAML. Test it against last month’s data.
The Delta table — create security.detection.rules and security.detection.alerts. That’s your foundation.
A single Workflow — one scheduled job that executes active rules. Start with hourly, tune later.
The App — deploy the Gradio app. Give your SOC lead the URL. Watch them stop asking “what rules do we have?”

The framework grows organically. Once one rule works, the second is trivial. By the tenth rule, your team is writing rules faster than they ever did in a SIEM.

Links:

Author

Mariusz Derela

Cyber Security Specialist | DevSecOps | AI/ML

Why SIEM-style detection doesn’t scale #

The framework #

1. Rule format #

Why YAML + SQL #

MITRE ATT&CK mapping #

2. Testing rules against historical data #

Test format #

Test runner #

Running tests in CI #

3. Deployment pipeline #

Rule catalog table #

Deploy notebook #

Scheduled execution #

4. The Databricks App #

App structure #

app.yaml #

The app #

What changes in your SOC #

Getting started #

Related