Slipstream 5.1 Unveiled: New CEP, Session Windows & Event‑Driven Engine

Complex Event Processing (CEP)

Slipstream 5.1 enriches the semantics and functions of CEP. CEP detects patterns of events in real‑time streams, allowing users to define complex event patterns composed of multiple events.

Key concepts:

Event : an occurrence or state change.

Operation : the logical order and lifecycle control of events.

Example: Detect a situation where the count (cnt) of objects of the same product model increases by more than 2, and this occurs exactly twice within 10 minutes. The following SQL expresses this logic and stores matches in times_tb1:

INSERT INTO times_tb1
SELECT e1.id, e1.cnt
FROM PATTERN (
  e1=stream_d1 [e1.cnt > 0]
  FOLLOWEDBY
  e2=stream_d1 [e2.id = e1.id and e2.cnt-e1.cnt > 2][TIMES(2)]
)
WITHIN ('10' minute)

Slipstream also supports additional pattern operators such as ONEORMORE, NEXT, NOTNEXT, NOTFOLLOWEDBY, etc., and provides features like real‑time processing, advanced analysis, support for fixed and infinite windows, and various matching modes.

Session Window

Slipstream 5.1 adds Session Window support, which groups events based on inactivity gaps, creating separate windows for distinct user sessions.

SQL example for analyzing data within a session window:

INSERT INTO times_tb1
SELECT id, sum(value)
FROM source
SESSIONWINDOW(
  INIT[op="login"]
  END[op="logout" EXCLUDE]
  KEYEDBY(id)
  EXPIRE('60' second DISCARD)
)
GROUP BY id;

This feature is useful for scenarios like website user behavior analysis, where actions naturally form sessions.

Event‑Driven Engine (Morphling Engine)

The event‑driven engine processes streams record‑by‑record, offering lower latency than mini‑batch processing. Slipstream 5.1 enhances this engine by allowing custom data sources and sinks, including writing to partitioned or bucketed tables.

Example of creating a custom HBase sink:

CREATE TABLE write_hbase_tb(
  id string, ts TIMESTAMP, url string, pv int)
TBLPROPERTIES(
  "output"="custom",
  "custom.writer.classpath"="io.transwarp.writer.vendor.cdh.CdhHBaseOutputFormat",
  "cdh.hbase.zookeeper.quorum"="172.16.2.23"
);

Example of defining a custom data source:

CREATE STREAM custom_stream(id string, name string, num int)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
TBLPROPERTIES(
  "source"="custom",
  "slipstream.datasource.classpath"="io.transwarp..source.RabbitMQCustomDataSource",
  "slipstream.datasource.queue"="test"
);

Additional enhancements include MERGE INTO support, window functions, a RESTful metrics API, and a History Server for job monitoring.

Conclusion

Slipstream 5.1 improves usability through three main advances: richer CEP syntax, the addition of Session Window analysis, and expanded event‑driven engine capabilities, making the platform more flexible and developer‑friendly for real‑time data flow.