Slipstream 5.1 -- 实时计算引擎多方面升级

Transwarp Slipstream是先进的实时流处理产品，不同于以往只允许通过Java或者Scala进行流应用开发的传统/开源实时计算引擎，Slipstream支持以SQL的方式实现业务逻辑，极大的降低了流应用开发的门槛，对普通业务人员也同样友好。

Slipstream提供丰富且全面的功能，例如复杂事件处理（CEP）、支持微批处理和事件驱动双模式，可同时服务于低延时和高吞吐两种实时计算场景，能够应对几乎所有类型的业务。同时，它还提供高可用性（HA）和Exactly-Once语义，为实现实时应用提供稳定、可靠的保障。

Slipstream 5.1在原有功能和性能的基础上进行近一步扩增和优化，主要反映在以下三个功能模块：复杂事件处理、会话窗口、事件驱动处理引擎。本文将详细解读Slipstream 5.1中这些具体变化。

复杂事件处理（CEP）

Slipstream 5.1对于原有复杂事件处理的语义以及功能进行了丰富。

复杂事件处理（CEP ：Complex Event Processing），是一种从实时事件流中检测符合特定匹配模式的一系列事件并进行处理、分析、匹配和挖掘的有效技术手段。

复杂事件处理包含两个关键概念：

• 事件：可以理解为一件事情，事物间的动作或者事物状态。

• 操作：事件的逻辑顺序及其生命周期的控制。

我们将这种通过操作组合在一起的一系列事件称为复杂事件模式，Slipstream可以处理由多个事件组成的复合事件，通过分析多个事件间的关系和模式对比从繁杂的事件流中找出“有意义”的信息（例如潜在的威胁或行为预测）。

假设需要找到符合这样一种模式的现象：同一个型号产品内的物体数量（cnt）增加量大于2，并且在10分钟内恰好出现了两次这种情况，则需要检测出来并判断为异常。

在Slipstream 5.1可以用如下的基于流的SQL语句表达上面的逻辑，并将命中模式的结果记录于表times_tb1中：

INSERT INTO times_tb1
SELECT e1.id, e1.cnt

-- 关键字PATTERN定义一个模式

FROM PATTERN (

-- 将事件e1定义为 cnt大于零的记录
e1=stream_d1 [e1.cnt > 0]

-- FOLLOWBY表示要求事件e2发生在e1之后
FOLLOWEDBY

-- 将事件e2定义为id与e1相同，cnt值与e1相差大于2，并恰好出现两次
e2=stream_d1 [e2.id = e1.id and e2.cnt-e1.cnt>2][TIMES(2)])

-- 时间窗口为10分钟

WITHIN ('10' minute)

除了示例中的语法，Slipstream还支持如ONEORMORE、NEXT、NOTNEXT、NOTFOLLOWEDBY等其他多种语法功能，用于充分的表达并满足各种业务对复杂模式的需求。

Slipstream 5.1不单单丰富了CEP的功能，提升了描述能力，而且还为其提供了以下特征：

• 实时性：一旦模式匹配，立即处理并生成结果。

• 分析能力：支持比较复杂的模式。

• 数据流：支持固定窗口和无限窗口数据流。

• 支持多种模式：单例、循环。

• 多种匹配方式：过滤，量化。

通过Slipstream，一般的非技术用户用SQL就可以开发基于实时数据的应用，应对例如股市趋势分析、信用卡反欺诈检测、IoT设备网络的监控、网络入侵检测等与复杂事件相关的应用场景。

会话窗口

Slipstream中新增了会话窗口（Session Window）功能。会话窗口是一种特殊的窗口，它会根据用户会话中非活跃的间隙将数据流分隔为多个窗口。同一时间间隔阈值下的消息聚集在同一个窗口内，超过时间阈值的消息会被分配到下一窗口。

一个会话窗口由会话的开始标记、结束标记、会话最长间隔三个元素来决定。Slipstream 5.1支持用SQL对会话窗口内的数据进行分析，下面是通过SQL实现分析的一个语句示例。

INSERT INTO times_tb1
SELECT id, sum(value)
FROM source

SESSIONWINDOW(

   -- INIT指定会话开始标记

   INIT[op="login"]

   -- END指定会话结束标记

   END[op="logout" EXCLUDE]

   -- 指定切分Session的字段

   KEYEDBY(id)

   -- 窗口的最长间隔为60秒

   EXPIRE('60' second DISCARD)

) GROUP BY id;

会话窗口功能将十分适用于如网站用户行为分析等具有明显会话特征的业务场景，因为用户在网站的不同活动通常带有“窗口”特性，例如浏览商品的行为在一个时间窗口，而购买商品的一系列行为可能会在另一个窗口。属于同一个会话窗口的消息关联性比较大，作为一组进行分析更有意义。Session Window功能恰好实现根据会话活动对元素进行分组，而且用SQL实现时间窗口可以让针对于用户行为的实时分析更加方便，更容易实现，对于调试更加友好。

事件驱动引擎

事件驱动引擎（Morphling Engine）是Slipstream从4.8开始引入的处理模式，对于流数据采用逐条读取记录，进行业务逻辑加工，再输出到终端的处理方式。相比于等待消息集合后再打包的微批（mini-batch）处理模式，事件驱动的流处理延迟更低，在对延迟敏感的业务场景中表现更佳。

Slipstream 5.1对事件驱动引擎进行了新的功能丰富和改进。

首先从5.1起Slipstream开始支持用户自定义数据源和目标位置，并且支持数据输出到分区表/分桶表。

例如如下语句将Sliptream数据的目标存储自定义为HBase表：

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"

… )

如下语句在Slipstream中自定义一个数据源：

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",

… )

此外，Slipstream 5.1还支持MERGE INTO语义、窗口函数、提供RESTful接口查询Morphling Job的指标。并且提供History Server服务，用来存储Slipstream Job的历史记录，并提供监控和展示的接口。

结语

总体而言，Slipstream 5.1通过下面三方面的改进，向着更易用的流处理产品成长：

1. 丰富了复杂事件处理（CEP）的语法，描述能力有所提升。

2. 增加会话窗口功能，实现以会话行为组的分析。

3. 事件驱动模式的语法支持与功能得到进一步增强。

由于功能的丰富和更好的语法支持度，可以看到Slipstream 5.1的使用性比以往的版本都更加友好，具备更高的灵活度，使得开发过程更方便，数据流动更加通畅。

---

往期原创文章

大数据开放实验室由星环信息科技（上海）有限公司运营，专门致力于大数据技术的研究和传播。若转载请在文章开头明显注明“文章来源于微信订阅号——大数据开放实验室”，并保留作者和账号介绍。