Lakewatch Sees Threats. Who Sees the Blind Spots?

On March 24, Databricks announced Lakewatch - a SIEM built natively on the Lakehouse. Detection rules, investigation workflows, threat intelligence, all running on Delta tables and Unity Catalog. For anyone who’s been building security analytics on Databricks, this isn’t surprising. It’s confirmation. The Security Data Lake isn’t a cheaper SIEM alternative anymore. It’s becoming the platform.

This is the direction I’ve been pushing for a while. My threat hunting app runs on Databricks. My detection-as-code framework stores rules in YAML and tests them against Delta tables. The data pipeline I built for this demo uses Zerobus Ingest through NiFi to land security telemetry directly into Delta.

Lakewatch will handle detection. Spark Declarative Pipelines keep the plumbing healthy. But there’s a gap between those two that neither is designed to fill.

The gap nobody talks about

#

Databricks already gives you solid observability over your data pipelines. SDP data quality expectations will tell you when a column has unexpected NULLs. You’ll get notified when a pipeline fails entirely. Schema drift gets caught before bad data hits your gold tables.

All of that works well for pipeline health. The problem is a different kind of failure.

You have a thousand endpoints pushing security telemetry through your ingestion pipeline. One of them stops reporting. Maybe an EDR agent crashed. Maybe someone rotated a credential and the agent can’t authenticate. Maybe a firewall rule changed and the host can’t reach the collector.

The pipeline is green. Data quality checks pass. Throughput barely dips - 999 out of 1,000 sources are still reporting. SDP expectations are met. Lakewatch detects threats in the data that arrives. Everything looks healthy.

Except one endpoint is now completely unmonitored, and nobody noticed for three days.

This isn’t a pipeline problem. It’s a visibility problem. The pipeline works fine. The data that arrives is valid. The issue is what’s not arriving.

Pipeline monitoring tells you the pipe works. Asset-level monitoring tells you nothing is missing.

SilentPulse

#

This is what I’ve been building with SilentPulse. It monitors security telemetry at the asset level - per host, per feed, per source. It compares what should be reporting against what actually is. When a source goes silent, SilentPulse catches it.

Over the past few weeks, I connected it natively to Databricks. Not as a separate tool that sends webhook alerts, but as a first-class data source that plugs into the Lakehouse ecosystem.

The integration has three packages:

• silentpulse-datasource - a PySpark DataSource that lets you read SilentPulse data with spark.read.format("silentpulse")
• silentpulse-sdp - observability decorators for Spark Declarative Pipelines
• silentpulse-uc-discovery - Unity Catalog asset discovery

All open source, Apache 2.0: github.com/silentpulse-io/silentpulse-databricks

The rest of this post is a full walkthrough. Real pipeline, real data, real failure simulation.

The demo environment

#

I’m using the same NiFi + Zerobus Ingest pipeline from my previous post. Four hosts (host1, host2, host3a, host3b) generate security telemetry continuously. NiFi routes it through MergeContent and PutZerobusIngest into a Delta table at demo.default.zerobus_ingestion.

NiFi (host1, host2, host3a, host3b)
  -> MergeContent
  -> PutZerobusIngest
  -> Databricks (demo.default.zerobus_ingestion)

SilentPulse monitors all four hosts. If any of them stops reporting, SilentPulse creates a silence alert.

Before anything else, verify the data is flowing:

1
2
3
4

SELECT host, COUNT(*) as events, MAX(timestamp) as last_seen
FROM demo.default.zerobus_ingestion
GROUP BY host
ORDER BY host

Four hosts, all reporting. This is our baseline.

Part 1: Reading SilentPulse data in Spark

#

The DataSource package installs from a wheel stored in a Unity Catalog Volume:

1
2

%pip install /Volumes/silentpulse/monitoring/packages/silentpulse_datasource-0.1.7-py3-none-any.whl
dbutils.library.restartPython()

Register the data source and connect:

1
2
3
4
5

from silentpulse_datasource import SilentPulseDataSource
spark.dataSource.register(SilentPulseDataSource)

SP_URL = dbutils.secrets.get("silentpulse", "api_url")
SP_TOKEN = dbutils.secrets.get("silentpulse", "api_token")

Now read alerts like any other data source:

1
2
3
4
5
6
7
8

df_alerts = (
    spark.read.format("silentpulse")
    .option("url", SP_URL)
    .option("token", SP_TOKEN)
    .option("resource", "alerts")
    .load()
)
df_alerts.display()

One line of code. The DataSource handles pagination, authentication, retry with exponential backoff on transient errors, and schema mapping. Under the hood it’s a full PySpark DataSource V2 implementation with batch reader, stream reader (for alerts), and stream writer (for telemetry ingest). The batch reader paginates through the SilentPulse API, yielding rows that map directly to the DataFrame schema.

Same pattern for pipeline coverage:

1
2
3
4
5
6
7
8

df_coverage = (
    spark.read.format("silentpulse")
    .option("url", SP_URL)
    .option("token", SP_TOKEN)
    .option("resource", "coverage")
    .load()
)
df_coverage.display()

Five resources are available: alerts, flows, entities, observations, and coverage. Each maps to a SilentPulse API endpoint with a predefined schema.

Materializing to Delta

#

Read once, write to Unity Catalog, query forever:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

for resource in ["alerts", "flows", "coverage"]:
    df = (
        spark.read.format("silentpulse")
        .option("url", SP_URL)
        .option("token", SP_TOKEN)
        .option("resource", resource)
        .load()
    )
    df.write.mode("overwrite").saveAsTable(f"silentpulse.monitoring.{resource}")
    print(f"Saved {resource}: {df.count()} rows")

Now the visibility data sits alongside your security telemetry in Unity Catalog. Same governance, same access controls, same query engine.

Joining both worlds

#

This is where it gets interesting. The raw events from Zerobus and the visibility metadata from SilentPulse, side by side:

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

WITH host_activity AS (
    SELECT
        host,
        COUNT(*) as total_events,
        MAX(timestamp) as last_event
    FROM demo.default.zerobus_ingestion
    GROUP BY host
)
SELECT
    h.host,
    h.total_events,
    h.last_event,
    a.status as alert_status,
    a.severity,
    a.started_at as alert_since
FROM host_activity h
LEFT JOIN silentpulse.monitoring.alerts a
    ON h.host = a.asset_identifier
    AND a.status = 'active'
ORDER BY h.host

One query that shows both what your hosts are sending and whether SilentPulse has flagged any of them. If a host has events but also an active alert, something changed recently. If a host has no events and an active alert, it went dark.

Part 2: Breaking things on purpose

#

Theory is nice. Let’s break something.

In NiFi, I stop the processor for host3b. The other three hosts keep generating events. The pipeline stays green. Zerobus keeps ingesting. Data quality checks pass.

After a few minutes, SilentPulse detects the silence. A new alert appears in the SilentPulse UI:

Back in Databricks, refresh the data:

1
2
3
4
5
6
7
8

df_alerts_new = (
    spark.read.format("silentpulse")
    .option("url", SP_URL)
    .option("token", SP_TOKEN)
    .option("resource", "alerts")
    .load()
)
df_alerts_new.filter("status = 'active'").display()

host3b: status active, alert_type silence. The pipeline never noticed. SilentPulse did.

Part 3: Pipeline observability with SDP decorators

#

The DataSource reads from SilentPulse. The SDP package goes the other direction - it reports telemetry back to SilentPulse from your Spark pipelines.

1
2

%pip install /Volumes/silentpulse/monitoring/packages/silentpulse_sdp-0.1.7-py3-none-any.whl
dbutils.library.restartPython()

1
2
3
4
5
6

import silentpulse_sdp

SP_URL = dbutils.secrets.get("silentpulse", "api_url")
SP_TOKEN = dbutils.secrets.get("silentpulse", "api_token")

silentpulse_sdp.configure(api_url=SP_URL, api_token=SP_TOKEN)

Three decorators, one function:

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

from silentpulse_sdp import heartbeat, volume, completeness

@heartbeat(integration_point="zerobus-ingest", interval_seconds=300)
@volume(integration_point="zerobus-ingest", min_rows=10)
@completeness(
    integration_point="zerobus-ingest",
    expected_columns=["host", "timestamp"],
)
def monitored_zerobus_ingest():
    return spark.table("demo.default.zerobus_ingestion")

result = monitored_zerobus_ingest()
print(f"ZeroBus ingest: {result.count()} rows from "
      f"{result.select('host').distinct().count()} hosts")

Each decorator is non-blocking. They fire telemetry to SilentPulse’s ingest endpoint using a thread pool, so your pipeline doesn’t slow down. The decorators report:

• heartbeat - “this pipeline is alive and ran at this time”
• volume - “this run processed N rows” (alerts if below min_rows)
• completeness - “these columns exist and are not all null”

Now SilentPulse has visibility from both directions. It monitors whether sources are reporting (asset-level), and it receives heartbeats from your Databricks pipelines that process that data. If either side goes silent, you’ll know.

Streaming: foreachBatch telemetry

#

The decorators above work great for batch tables. But if your SDP table returns a streaming DataFrame, operations like collect() and groupBy() fail, so the decorators fall back to heartbeat-only. You lose volume, completeness, and freshness signals.

Since v0.1.7, the SDP package ships streaming_sink() and report_batch() for foreachBatch integration. Spark calls your handler with each micro-batch as a regular DataFrame, so all aggregation works:

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

from silentpulse_sdp import streaming_sink

sink = streaming_sink(
    integration_point="zerobus-ingest",
    asset_column="host",
    min_rows=10,
    max_delay_seconds=600,
)

@dlt.table(name="telemetry_sink")
def telemetry_sink():
    return (
        spark.readStream.table("bronze")
        .writeStream.foreachBatch(sink)
    )

Each micro-batch reports heartbeat + volume per asset. Add expected_columns for completeness checks, max_delay_seconds for freshness. If you need to write the batch to a table alongside telemetry, use report_batch() inside your own handler:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from silentpulse_sdp import report_batch

def my_sink(batch_df, batch_id):
    report_batch(
        batch_df, batch_id,
        integration_point="zerobus-ingest",
        asset_column="host",
        min_rows=10,
    )
    batch_df.write.mode("append").saveAsTable("silver_events")

Everything is exception-safe. If SilentPulse is unreachable or the DataFrame is empty, the pipeline continues without interruption.

Full SDP pipeline with foreachBatch sink

#

Putting it all together - a complete Spark Declarative Pipeline that reads from Zerobus, builds a gold materialized view, and reports telemetry to SilentPulse in parallel:

 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

from pyspark import pipelines as dp
from pyspark.sql import functions as F
from silentpulse_sdp import report_batch

import silentpulse_sdp
silentpulse_sdp.configure(
    api_url=dbutils.secrets.get("silentpulse", "api_url"),
    api_token=dbutils.secrets.get("silentpulse", "api_token"),
)

# Source streaming table
@dp.table(name="zerobus_ingestion")
def zerobus_ingestion():
    return spark.readStream.table("demo.default.zerobus_ingestion")

# Gold materialized view - events per host per hour
@dp.materialized_view(name="gold_host_hourly")
def gold_host_hourly():
    return (
        spark.read.table("zerobus_ingestion")
        .groupBy(
            "host",
            F.date_trunc("hour", (F.col("timestamp") / 1000).cast("timestamp")).alias("hour"),
        )
        .agg(
            F.count("*").alias("event_count"),
            F.max("timestamp").alias("last_seen"),
        )
    )

# SilentPulse telemetry sink
@dp.foreach_batch_sink(name="silentpulse_sink")
def silentpulse_sink(batch_df, batch_id):
    from silentpulse_sdp import report_batch
    report_batch(
        batch_df, batch_id,
        integration_point="zerobus-ingest",
        asset_column="host",
        min_rows=5,
        expected_columns=["host", "timestamp"],
        max_delay_seconds=600,
    )

@dp.append_flow(target="silentpulse_sink")
def silentpulse_flow():
    return spark.readStream.table("zerobus_ingestion")

The pipeline DAG shows the three components: streaming table feeding both the gold MV and the SilentPulse sink in parallel:

The foreach_batch_sink is a first-class SDP construct. It handles checkpointing, retries, and lifecycle automatically. Each micro-batch flows through report_batch() which sends per-asset heartbeat, volume, completeness, and freshness telemetry to SilentPulse without blocking the pipeline.

Part 4: Metric Views for visibility KPIs

#

Raw data is useful. Governed metrics are better. Unity Catalog Metric Views (available on DBR 17.2+) let you define reusable measures in YAML and query them with MEASURE().

I built two metric views for SilentPulse data. The SQL to deploy them:

 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

CREATE OR REPLACE VIEW silentpulse.monitoring.alert_metrics
WITH METRICS
LANGUAGE YAML
AS $$
version: 1.1
comment: "Security telemetry visibility metrics from SilentPulse alerts"
source: silentpulse.monitoring.alerts

dimensions:
  - name: Severity
    expr: severity
  - name: Alert Type
    expr: alert_type
  - name: Asset Group
    expr: asset_group_name
  - name: Flow Name
    expr: flow_name
  - name: Alert Day
    expr: DATE_TRUNC('DAY', started_at)
  - name: Alert Week
    expr: DATE_TRUNC('WEEK', started_at)

measures:
  - name: Active Alerts
    expr: COUNT(CASE WHEN status = 'active' THEN 1 END)
  - name: Silent Feeds
    expr: COUNT(CASE WHEN status = 'active' AND alert_type = 'silence' THEN 1 END)
  - name: Blind Spot Hours
    expr: SUM(TIMESTAMPDIFF(HOUR, started_at, COALESCE(resolved_at, current_timestamp())))
  - name: Mean Time to Detect (min)
    expr: AVG(TIMESTAMPDIFF(MINUTE, started_at, resolved_at))
  - name: Max Gap Duration (hours)
    expr: MAX(TIMESTAMPDIFF(HOUR, started_at, COALESCE(resolved_at, current_timestamp())))
$$;

The metric view separates what you measure from how you group it. The same Blind Spot Hours measure works whether you group by severity, by asset group, by week, or by all three at once.

Query it:

1
2
3
4
5
6
7
8
9

SELECT
    `Severity`,
    `Asset Group`,
    MEASURE(`Active Alerts`) AS active_alerts,
    MEASURE(`Silent Feeds`) AS silent_feeds,
    MEASURE(`Blind Spot Hours`) AS blind_spot_hours
FROM silentpulse.monitoring.alert_metrics
GROUP BY ALL
ORDER BY active_alerts DESC

A second metric view covers feed health with measures like Total Assets and Coverage Pct, using the same pattern against silentpulse.monitoring.coverage.

These metrics are first-class objects in Unity Catalog. They show up in Catalog Explorer with their dimensions and measures listed. They work natively with AI/BI Dashboards and Genie Spaces. And because they’re governed by UC, access controls apply automatically.

Part 5: AI-powered root cause analysis

#

An alert tells you that host3b went silent. It doesn’t tell you why. Was it a network outage? A credential rotation? A pipeline failure? A decommissioned host?

Databricks AI Functions answer that question with one SQL call. No model to deploy, no endpoint to manage, no GPU to provision.

 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

SELECT
    id,
    asset_identifier,
    flow_name,
    severity,
    ai_classify(
        CONCAT(
            'Security feed "', flow_name,
            '" monitoring host "', asset_identifier,
            '" went silent at ', CAST(started_at AS STRING),
            '. This host was sending telemetry via NiFi -> Kafka -> ZeroBus pipeline.',
            ' Alert type: ', alert_type,
            '. Severity: ', severity, '.'
        ),
        ARRAY(
            'network_outage',
            'credential_rotation',
            'source_decommission',
            'pipeline_failure',
            'rate_limiting',
            'maintenance_window'
        )
    ) AS probable_cause
FROM silentpulse.monitoring.alerts
WHERE status = 'active'
ORDER BY started_at DESC

ai_classify takes a text description and a set of labels, and returns the most likely match. For host3b, which we know stopped because we killed the NiFi processor, it returns pipeline_failure. No model training. No prompt engineering. One function call.

For an executive summary across all active alerts:

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

WITH alert_context AS (
    SELECT CONCAT_WS('\n',
        CONCAT('Total active alerts: ', COUNT(*)),
        CONCAT('High severity: ', COUNT(CASE WHEN severity = 'high' THEN 1 END)),
        CONCAT('Affected hosts: ',
            ARRAY_JOIN(COLLECT_SET(asset_identifier), ', ')),
        CONCAT('Affected flows: ',
            ARRAY_JOIN(COLLECT_SET(flow_name), ', ')),
        'These hosts send security telemetry via NiFi -> Kafka -> ZeroBus to Databricks.',
        'SilentPulse monitors whether each host is reporting as expected.'
    ) AS context
    FROM silentpulse.monitoring.alerts
    WHERE status = 'active'
)
SELECT ai_summarize(context, 50) AS executive_summary
FROM alert_context

One paragraph that your CISO can read. Generated from live data, not a static report.

Persisting enriched data

#

For dashboards and automation, materialize the AI-enriched alerts:

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

CREATE OR REPLACE TABLE silentpulse.monitoring.alerts_enriched AS
SELECT
    a.*,
    ai_classify(
        CONCAT('Host "', asset_identifier, '" on flow "', flow_name,
            '" silent since ', CAST(started_at AS STRING),
            '. NiFi/Kafka/ZeroBus pipeline. Type: ', alert_type, '.'),
        ARRAY('network_outage', 'credential_rotation', 'source_decommission',
            'pipeline_failure', 'rate_limiting', 'maintenance_window')
    ) AS probable_cause
FROM silentpulse.monitoring.alerts a
WHERE status = 'active'

Now you have a Delta table with AI-classified root causes that updates on every refresh. Dashboards, alerts, notebooks - all can query it directly.

Part 6: The dashboard

#

Everything above feeds into an AI/BI Dashboard. KPI counters at the top, charts in the middle, detail tables at the bottom. All querying the same Delta tables and metric views.

Row 1 - KPI cards:

• Active Alerts (counter)
• Blind Spot Hours (counter)
• Affected Hosts (counter)

Row 2 - Charts:

• Root Cause Distribution (pie chart from alerts_enriched)
• Severity Breakdown by Cause (stacked bar)

Row 3 - Detail:

• Top Affected Hosts (table with event counts and alert status)
• AI Insights with recommended actions

The dashboard is backed by six SQL queries. The full query set is in the repo at examples/dashboards/ai-enriched-alerts.sql.

Part 7: Natural language queries with Genie

#

The final layer. A Genie Space configured with the SilentPulse tables and metric views, where anyone can ask questions in plain English.

 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

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

space = w.genie.create_space(
    title="SilentPulse Security Visibility",
    description=(
        "Ask questions about security telemetry visibility, alert trends, "
        "feed health, and detection coverage monitored by SilentPulse."
    ),
    warehouse_id=WAREHOUSE_ID,
    table_identifiers=[
        "silentpulse.monitoring.alerts",
        "silentpulse.monitoring.flows",
        "silentpulse.monitoring.coverage",
        "silentpulse.monitoring.alert_metrics",
        "silentpulse.monitoring.feed_health_metrics",
    ],
    sample_questions=[
        "Which security feeds went silent in the last 24 hours?",
        "What is our current detection coverage percentage?",
        "How many active alerts are there by severity?",
        "Which asset groups have the most blind spot hours?",
    ],
)

Ask “Which security feeds went silent in the last 24 hours?” and Genie translates it to SQL, returns a result table with flow names, asset groups and alert counts, and even generates a bar chart breakdown:

The metric views make Genie’s answers more accurate because the metrics are pre-defined. Genie doesn’t have to guess how to calculate “blind spot hours” - it uses MEASURE() on the metric view directly.

The complete picture

#

Here’s what the full stack looks like:

flowchart TB
    subgraph Sources["Security Telemetry"]
        H1["host1"] & H2["host2"] & H3["host3a"] & H4["host3b"]
    end

    subgraph NiFi["Apache NiFi"]
        MC["MergeContent"] --> ZB["PutZerobusIngest"]
    end

    subgraph Databricks["Databricks Lakehouse"]
        Delta["demo.default.zerobus_ingestion\n(Delta Table)"]
        SP_Tables["silentpulse.monitoring.*\n(Alerts, Flows, Coverage)"]
        MV["Metric Views\n(Alert KPIs, Feed Health)"]
        AI["AI Functions\n(ai_classify, ai_summarize)"]
        Dash["AI/BI Dashboard"]
        Genie["Genie Space"]
    end

    subgraph SilentPulse["SilentPulse"]
        Monitor["Asset-level\nMonitoring"]
        API["REST API"]
    end

    Sources --> MC
    ZB --> Delta
    Monitor -->|"silence detection"| API
    API -->|"spark.read.format('silentpulse')"| SP_Tables
    SP_Tables --> MV
    SP_Tables --> AI
    AI --> Dash
    MV --> Dash
    MV --> Genie

    Delta -.->|"SDP decorators + foreachBatch\n(heartbeat, volume, freshness)"| API

    style Monitor fill:#6366f1,color:#fff
    style AI fill:#6366f1,color:#fff
    style MV fill:#10b981,color:#fff

Three monitoring layers, each catching different problems:

SDP and Lakewatch work on data that arrives. SilentPulse watches for data that doesn’t.

Running the demo yourself

#

Everything you need:

1. Databricks workspace with DBR 15.4+ and a Pro/Serverless SQL Warehouse (17.2+ for Metric Views)
2. SilentPulse instance monitoring your data sources (or the hosted demo)
3. Wheels from GitHub Release v0.1.7 uploaded to a UC Volume

1
2
3

CREATE CATALOG IF NOT EXISTS silentpulse;
CREATE SCHEMA IF NOT EXISTS silentpulse.monitoring;
CREATE VOLUME IF NOT EXISTS silentpulse.monitoring.packages;

Upload the three wheels, install them, and follow the notebooks in examples/. The metric view SQL is at examples/metric-views/deploy-metric-views.sql. The dashboard queries are at examples/dashboards/ai-enriched-alerts.sql.

The full repo: github.com/silentpulse-io/silentpulse-databricks

What’s next

#

The SDP package now supports both batch decorators and streaming via foreachBatch, so you get full per-asset telemetry regardless of how your pipeline runs. The DataSource handles batch reads and streaming reads for alerts, plus streaming writes for telemetry ingest.

UC Discovery is the next piece - automatically scanning Unity Catalog for tables that should have asset-level monitoring and suggesting SilentPulse configurations. If you have 200 Delta tables with a host or source_ip column, you probably want to know when one of those sources stops appearing.

The Databricks ecosystem keeps getting deeper. Lakewatch for detection. SDP for pipeline health. AI Functions for classification. Metric Views for governed KPIs. Genie for natural language access. The integration points are there. The missing piece was someone watching whether the sources feeding all of this are actually reporting.

That’s SilentPulse. See when security goes silent.

Links:

• silentpulse-databricks - Integration repo (Apache 2.0)
• SilentPulse - Project homepage
• Databricks Lakewatch announcement
• Zerobus Ingest + NiFi - Previous post on the data pipeline
• Metric Views documentation
• AI Functions documentation