Unlock Complex Event Processing with Transwarp Slipstream: Real‑World Examples

Transwarp Slipstream is the stream‑processing component of Transwarp Data Hub (TDH) that allows developers to build real‑time applications using only SQL.

Complex Event Processing (CEP)

CEP is a technique for detecting sequences of events that match a defined pattern in a continuous stream, enabling analysis, mining, and timely response to significant situations such as threats or behavior predictions.

Enabling CEP

To activate CEP, set the ngmr.engine.mode parameter to morphling in Transwarp Manager, save the configuration, and restart Slipstream.

After saving, execute the "Configure Service" and "Restart Slipstream" steps.

CEP Basic Syntax

The core syntax follows the pattern:

SELECT EVENT1.column, EVENT2.column, ...
FROM PATTERN (
  EVENT1=[stream][condition] [FOLLOWEDBY|,]
  EVENT2=[stream][condition] [FOLLOWEDBY|,]
  ...
) WITHIN (time interval);

Where:

EVENT : user‑defined name of each event in the pattern.

PATTERN : keyword that introduces the pattern definition.

FOLLOWEDBY : indicates the second event must occur after the first; a comma (,) means it must be the immediate next event.

condition : SQL‑style filter applied to the event.

WITHIN : time window for the whole pattern.

Application Example

Scenario 1: Bank fraud detection

The bank needs to detect possible card‑skimming fraud within ten minutes of a withdrawal.

CREATE APPLICATION cep_example;
USE APPLICATION cep_example;
SET streamsql.use.eventmode=true;
CREATE STREAM transaction(
  location_id STRING, card_id STRING, behavior STRING
) tblproperties(
  "topic"="transaction_t1",
  "kafka.zookeeper"="localhost:2188",
  "kafka.broker.list"="localhost:9098"
);
CREATE TABLE exception_ret(
  location_id_1 STRING,
  location_id_2 STRING,
  behavior STRING,
  card_id STRING
);

INSERT INTO exception_ret
SELECT e1.location_id, e1.card_id, e1.behavior,
       e2.location_id, e2.card_id, e2.behavior
FROM PATTERN (
  e1=transaction[e1.behavior='withdraw']
  FOLLOWEDBY
  e2=transaction[
    e2.card_id = e1.card_id AND
    e2.behavior='withdraw' AND
    e2.location_id != e1.location_id]
) WITHIN ('10' minute);

Scenario 2: Traffic flow and cloned‑vehicle monitoring

Two cases are addressed: (1) alert when traffic volume at a checkpoint exceeds a threshold within ten minutes, and (2) detect cloned plates appearing in different regions within ten minutes.

CREATE APPLICATION cep_example;
USE APPLICATION cep_example;
SET streamsql.use.eventmode=true;
CREATE STREAM traffic(
  veh_id STRING, veh_type STRING, speed FLOAT, location_id STRING
) tblproperties(
  "topic"="traffic_t1",
  "kafka.zookeeper"="localhost:2188",
  "kafka.broker.list"="localhost:9098"
);
CREATE TABLE traffic_flow_ret(location_id STRING, traffic_flow INT);
CREATE TABLE traffic_susp(loc_id1 STRING, loc_id2 STRING, veh_id STRING);

CREATE STREAM traffic_flow AS
SELECT location_id, count(*) as veh_flow
FROM traffic STREAMWINDOW w1 as (length '1' minute slide '1' minute)
GROUP BY location_id;

INSERT INTO traffic_flow_ret
SELECT e2.location_id, e2.veh_flow
FROM PATTERN(
  e1=traffic_flow[e1.veh_flow > 0],
  e2=traffic_flow[
    e2.veh_flow - e1.veh_flow > 60 AND
    e2.location_id = e1.location_id]
) WITHIN ('10' minute);

INSERT INTO traffic_susp
SELECT e1.location_id, e2.location_id, e2.veh_id
FROM PATTERN(
  e1=traffic[veh_type="A1"] FOLLOWEDBY
  e2=traffic[veh_id = e1.veh_id AND e2.location_id != e1.location_id]
) WITHIN ('10' minute);

Conclusion

Transwarp Slipstream empowers users to perform comprehensive real‑time data mining on streams, supporting both event‑driven and micro‑batch modes, and offering advanced CEP capabilities that enable timely decision‑making when complex event patterns emerge.