Scenario D: Security events¶

Part of Streaming Transformations › Scenarios.

Goal¶

Security teams ingest large event streams, but only a few event types matter. Each relevant event needs context about the asset and any known threat, a normalized severity, and one row per owning team so that alerts reach the right people.

Transformations used¶

Stream filtering on stream and dimension columns: keep relevant event types, and keep events whose threat-intelligence risk score is at or above a threshold.
Stream enrichment: join the asset inventory and threat-intelligence tables, two dimension joins in one pipe.
Custom scalar logic: a SQL UDF classifies the severity of each event.
Lateral Flatten on the dimension: expand the asset’s array of owning teams after the join, so each team gets its own row.

Source data¶

Stream¶

The example uses these five records:

{"ts": "2026-10-09 08:00:01", "event_type": "auth_failure", "user": "bob", "src_ip": "203.0.113.7", "host": "db-prod-01"}
{"ts": "2026-10-09 08:00:02", "event_type": "login_success", "user": "alice", "src_ip": "203.0.113.7", "host": "db-prod-01"}
{"ts": "2026-10-09 08:00:03", "event_type": "privilege_escalation", "user": "carol", "src_ip": "198.51.100.20", "host": "web-01"}
{"ts": "2026-10-09 08:00:04", "event_type": "auth_failure", "user": "dave", "src_ip": "192.0.2.10", "host": "web-01"}
{"ts": "2026-10-09 08:00:05", "event_type": "auth_failure", "user": "erin", "src_ip": "203.0.113.7", "host": "laptop-77"}

Dimension tables¶

The PIPE joins these dimension tables, refreshed periodically:

-- Threat-intelligence dimension: risk score and country per address.
CREATE OR REPLACE TABLE dim_ip_reputation (
  ip STRING PRIMARY KEY RELY,
  risk_score NUMBER,
  country STRING);

INSERT INTO dim_ip_reputation VALUES
  ('203.0.113.7', 90, 'XX'),
  ('198.51.100.20', 85, 'DE'),
  ('192.0.2.10', 20, 'US');

-- Asset inventory dimension: criticality and owning teams (an array) per host.
CREATE OR REPLACE TABLE dim_assets (
  host STRING PRIMARY KEY RELY,
  criticality STRING,
  owner_teams ARRAY);

INSERT INTO dim_assets
  SELECT 'db-prod-01', 'critical', ARRAY_CONSTRUCT('dba', 'secops') UNION ALL
  SELECT 'web-01', 'high', ARRAY_CONSTRUCT('web-platform');

Target table¶

The PIPE lands transformed and filtered rows in this target table:

CREATE OR REPLACE TABLE sec_events (
  event_ts TIMESTAMP_NTZ, event_type STRING, user_name STRING, src_ip STRING,
  host STRING, risk_score NUMBER, ip_country STRING, criticality STRING,
  severity STRING, owner_team STRING);

Transformation¶

The PIPE lands data in the target table. Events are enriched with a criticality level per asset and a risk score per IP address, joined from the dimension tables. A custom SQL UDF classifies the severity of each event based on the risk score and criticality. Relevant events are kept by filtering on the event type stream column, and the risk score dimension column.

-- SQL UDF: severity from the IP risk score and the asset criticality.
CREATE OR REPLACE FUNCTION classify_severity(risk_score NUMBER, criticality STRING)
  RETURNS STRING AS
  $$ CASE WHEN risk_score >= 90 AND criticality = 'critical' THEN 'critical'
          WHEN risk_score >= 80 THEN 'high'
          ELSE 'medium' END $$;

-- Two dimension joins in one pipe, filter on stream (event_type) and dimension
-- (risk_score) columns, classify with a SQL UDF, then flatten the asset's
-- owner_teams after the join; OUTER => TRUE keeps events on unknown assets.
CREATE OR REPLACE PIPE sec_events_pipe AS
  COPY INTO sec_events (event_ts, event_type, user_name, src_ip, host, risk_score, ip_country, criticality, severity, owner_team) FROM (
    SELECT
      e.$1:ts::TIMESTAMP_NTZ,
      e.$1:event_type::STRING,
      e.$1:user::STRING,
      e.$1:src_ip::STRING,
      e.$1:host::STRING,
      r.risk_score,
      r.country,
      a.criticality,
      classify_severity(r.risk_score, a.criticality),
      t.value::STRING
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) e
    LEFT OUTER JOIN dim_ip_reputation r
      ON e.$1:src_ip::STRING = r.ip
    LEFT OUTER JOIN dim_assets a
      ON e.$1:host::STRING = a.host,
         LATERAL FLATTEN(input => a.owner_teams, OUTER => TRUE) t
    WHERE e.$1:event_type::STRING IN ('auth_failure', 'privilege_escalation')
      AND r.risk_score >= 70);

Result¶

Querying the target table returns these rows:

EVENT_TS	EVENT_TYPE	USER_NAME	SRC_IP	HOST	RISK_SCORE	IP_COUNTRY	CRITICALITY	SEVERITY	OWNER_TEAM
2026-10-09 08:00:01	auth_failure	bob	203.0.113.7	db-prod-01	90	XX	critical	critical	dba
2026-10-09 08:00:01	auth_failure	bob	203.0.113.7	db-prod-01	90	XX	critical	critical	secops
2026-10-09 08:00:03	privilege_escalation	carol	198.51.100.20	web-01	85	DE	high	high	web-platform
2026-10-09 08:00:05	auth_failure	erin	203.0.113.7	laptop-77	90	XX	NULL	high	NULL

The login_success event from alice is filtered out by event type, and dave’s event is dropped because the risk score of 192.0.2.10 is 20, below the threshold of 70. The host laptop-77 isn’t in dim_assets, so erin’s event lands once with NULL criticality and owner team, and its severity is high because the critical branch requires a known critical asset.

Considerations¶

Dimension data is a snapshot that refreshes periodically, so updates can take several minutes to appear. This works well for slow-moving reference data. Dynamic tables can be used as dimension tables.
A pipe can join up to four dimension tables, and each table only once.
Use OUTER => TRUE on the dimension flatten. Otherwise events that match no asset are dropped.
ERROR_LOGGING currently can’t be enabled on the target table of a pipe that uses LATERAL FLATTEN.