WSO2流处理器相关学习（1）快速入门指南【转】

转至元数据结尾

转至元数据

快速入门指南

WSO2 流处理器 (SP) 是一个轻量级、精简、基于流 SQL 的流处理平台，允许您收集事件、实时分析它们、识别模式、映射它们的影响并在几毫秒内传达结果。它由Siddhi提供动力，具有极高的性能。

首先，让我们了解本指南中使用的以下概念：

流处理和复杂事件处理概述

让我们通过一个例子来理解什么是事件。如果我们将通过 ATM 进行的交易视为数据流，则可以将其取款视为一个事件。此事件包含有关金额、时间、帐号等的数据。许多此类交易形成一个流。

流处理引擎允许您创建处理图并将事件注入其中。每个操作员处理并将事件发送到下一个处理器。

复杂事件是总结、代表或表示一组其他事件的事件。复杂事件处理是流处理的一个子集，它涉及实时分析多个事件流，识别跨流的特定序列或模式，并从相关事件中推断出业务重要事件。

WSO2 SP 的流处理功能允许您捕获大量数据流并实时处理它们，并以流方式呈现结果，同时其复杂的事件处理功能通过 Siddhi 支持的模式和序列检测决策模式和趋势。

悉地概述

WSO2 SP 使用 Siddhi 查询语言为其 Siddhi 应用程序编写处理逻辑。悉地可以：

接受来自许多不同类型来源的事件输入
处理它们以产生见解
将它们发布到多种类型的接收器。

要使用Siddhi，您需要使用Siddhi Streaming SQL 语言将处理逻辑编写为Siddhi Application。在编写并启动 Siddhi 应用程序后，它：

将数据一一作为事件
处理每个事件中的数据
根据目前完成的处理生成新的高级事件
将新生成的事件作为输出发送到流。

在你开始之前：

安装Oracle Java SE 开发工具包 (JDK) 1.8 版。
设置 JAVA_HOME环境变量。
下载最新的WSO2 流处理器。
解压缩下载的 zip 并导航到<SP_HOME>/bin目录（<SP_HOME>是解压缩的目录）。
发出以下命令之一以启动 WSO2 Stream Processor Studio。
- 对于 Windows： editor.bat
- 对于 Linux：./editor.sh

成功启动 WSO2 SP 服务器后，将在 CLI 中打印类似于以下内容的日志。

服务器日志在启动日志中打印 Stream Processor Studio URL，如下所示。

设想

在这个场景中，您正在为一个虚构的大型航运公司 Shipping Wave 创建一个应用程序。史密斯先生，货物经理需要在任何给定时间跟踪装载到船上的货物的总重量。测量装载到船上的货箱重量被认为是一个事件。

让我们开始吧！您可以编写一个简单的 Siddhi 应用程序，按照以下步骤计算装载到船上的每个货箱的总重量。

第 1 步：创建 Siddhi 应用程序
第 2 步：模拟事件
第 3 步：编辑 Siddhi 应用程序以执行时间处理
步骤 4：为编辑的 Siddhi 应用程序模拟事件
下一步是什么？

第 1 步：创建 Siddhi 应用程序

Smith 需要计算每增加一个货箱后装载到船上的货物的总重量。为了为他生成这个输出，让我们创建一个 Siddhi 应用程序，如下所示：

通过http://<HOST_NAME>:<EDITOR_PORT>/editor URL访问 Stream Processor Studio 。

默认网址是 http://localhost:9390/editor

Stream Processor Studio 将打开，如下所示。
单击新建开始创建新的 Siddhi 应用程序。
输入您的 Siddhi 应用程序的名称。在这种情况下，让我们为应用程序命名 CargoWeightApp ，如下所示。

@App:name("CargoWeightApp")
定义输入流。流需要有一个名称和一个模式来定义每个传入事件应该包含的数据。事件数据属性表示为名称和类型对。在这个例子中：
- 输入流的名称： CargoStream
- 引用每个事件中数据的名称： weight
- 作为权重接收的数据类型： int
define stream CargoStream (weight int);

定义输出流。这与之前的定义具有相同的信息，但有一个附加totalWeight属性，其中包含迄今为止计算的总重量。在这里，我们需要添加一个sink配置来记录来自的事件，OutputStream以便我们可以观察输出值。

接收器指定通过 Siddhi 将流发布到外部系统的方法。在这种情况下，添加的接收器属于该log类型，它在 CLI 中将输出流发布为日志。

@sink(type='log', prefix='LOGGER')

define stream OutputStream(weight int, totalWeight long);

输入定义以下内容的 Siddhi 查询。

查询的名称（即， cargoWeightQuery）
从中获取要处理的事件的输入流（即， CargoStream）
需要发送到输出流的数据（即， weight 和 totalWeight）
需要如何计算输出（即通过计算所有事件的权重之和）
输出需要发送到的流（即， OutputStream）

此查询如下：

@info(name='CargoWeightQuery')

from CargoStream

select weight, sum(weight) as totalWeight

insert into OutputStream;

完成的Siddhi文件如下：

@App:name("CargoWeightApp")

define stream CargoStream (weight int);

@sink(type='log', prefix='LOGGER')

define stream OutputStream(weight int, totalWeight long);

@info(name='CargoWeightQuery')

from CargoStream

select weight, sum(weight) as totalWeight

insert into OutputStream;

要保存 Siddhi 文件，请单击 File => Save。这将打开 保存到工作区 对话框。单击保存将此文件保存在 <SP_HOME>/wso2/editor/deployment/workspace 目录中（这是保存 Siddhi 应用程序的默认位置）。

第 2 步：模拟事件

Stream Processor Studio 具有对模拟事件的内置支持。为了测试CargoWeightApp您创建的是否按预期工作，让我们按照下面给出的步骤模拟一些事件。

要启动 CargoWeight Siddhi 应用程序，请单击播放按钮。

如果应用程序启动成功，Stream Processor Studio 控制台会记录以下内容。
在 Stream Processor Studio 中单击以下图标以打开事件模拟面板。

在面板的 Single Simulation 选项卡中，选择如下值：

场地	价值
悉地应用名称	`CargoWeight`
流名称	`CargoStream`

结果，流的权重属性 CargoStream显示如下：

在权重字段中，输入 1000 并单击开始以启动 Siddhi 应用程序。然后单击发送以发送事件。该事件记录在 CLI 中，如下所示：

使用以下值再发送五个事件。

事件编号	重量
1	2000
2	1500
3	2000
4	3000
5	1000

事件记录如下。

每个新权重都添加到新权重中。因此，发送完所有六个事件后，总权重为10500。

第 3 步：编辑 Siddhi 应用程序以执行时间处理

本节演示如何使用 Siddhi 进行时间窗口处理。

在前面的场景中，您通过仅在内存中运行总和值来执行处理。在此过程中未存储任何事件。

窗口处理是一种方法，它允许我们在内存中存储给定时间段或特定数量的事件的一些事件，以便我们可以执行诸如计算其中的平均值、最大值等值的操作

让我们考虑一下，货物经理 Smith 有一个额外的要求，即每次装载新货箱时计算最后三个货箱的平均装载量，以平衡整个船的重量。在这里，我们正在考虑一个由三个事件组成的窗口，如下图所示。

为此，请按照以下步骤编辑 Siddhi 应用程序：

添加一个名为流averageWeight定义的新属性， OutputStream以便每个输出事件除了显示加载的新框的重量和总重量外，还显示平均重量。

define stream OutputStream(weight int, totalWeight long, averageWeight double);
要指定如何计算平均权重，请将 avgSiddhi 函数应用于语句中的 weight属性， select如下所示。这表明平均值是为weight传入事件的属性计算的。

select weight, sum(weight) as totalWeight, avg(weight) as averageWeight
要指定此查询对每个事件执行的计算必须仅应用于接收到的最后三个事件，请将长度窗口应用于输入流，如下所示。

from CargoStream#window.length(3)

此窗口适用于为从CargoStream流中获取的事件执行的所有计算。因此，添加此窗口还会导致根据最后三个事件计算总权重。

完成的查询如下。

@info(name='CargoWeightQuery')

from CargoStream#window.length(3)

select weight, sum(weight) as totalWeight, avg(weight) as averageWeight

insert into OutputStream;

完整的CargoWeightSiddhi 应用程序如下。

@App:name("CargoWeight")

define stream CargoStream (weight int);

@sink(type='log', prefix='LOGGER')

define stream OutputStream(weight int, totalWeight long, averageWeight double);

@info(name='CargoWeightQuery')

from CargoStream#window.length(3)

select weight, sum(weight) as totalWeight, avg(weight) as averageWeight

insert into OutputStream;

步骤 4：为编辑的 Siddhi 应用程序模拟事件

在这一步中，让我们启动编辑的 Siddhi 应用程序并模拟您在第 8 步中模拟的相同的六个事件。

事件编号	重量
事件编号	重量
1	1000
2	2000
3	1500
4	2000
5	3000
6	1000

生成的输出记录如下。

下一步是什么？

了解有关 Siddhi 的更多信息以及它如何在您的用户场景中提供帮助：教程

了解如何开发、部署和监控 Siddhi 应用程序：用户指南