操作步骤

1. 鼠标放在App下的Logic文件夹旁会出现加号，单击加号，选择“事件流”。

   您需要联系运维开启事件流License权限。开启后，才会有新建事件流的入口。

2. 单击“新建”，进入“事件流”页面。
3. 在“事件流”页面右侧单击“设置”，进行数据配置。
   图2 设置页面
   

   表1 参数说明

   参数名称	参数说明
   窗口长度	流处理的数据采集时间间隔，单位为分钟。建议时间间隔不要配置过小，否则系统频繁的建表和执行SQL语句，会影响系统性能。例如：配置为“1”表示spark streaming每隔1分钟去kafka获取事件数据。

4. 在“事件流”页面右侧单击“图元”，拖拽“输入源”中Kafka图标至左侧画布区域，并配置基本信息，单击“保存”。
   输入源中Kafka图标表示数据来自Kafka。

   图3 数据来源Kafka基本信息页面
   

   表2 数据来源Kafka基本信息页面参数说明

   参数名称	参数说明
   标签	该数据源Kafka的显示标签名。长度不超过64个字节。
   名称	该数据源Kafka的名称，系统自动生成。
   事件	需要订阅的事件。 例如：配置为温度监测事件“temperature”。
   从外部接收	是否从外部Kafka接收事件。 若关闭该开关，表示从与AppCube相连的Kafka接收事件；若打开该开关，表示从外部Kafka接收事件，需要配置“Kafka集群地址”和“订阅主题”。 Kafka集群地址：数据源Kafka集群地址。 多个节点服务器地址可用“,”间隔，格式为“Kafka节点1的ip:节点1端口号,Kafka节点2的ip:节点2端口号,...”。 获取Kafka的IP和端口号方法：可以通过登录每台Kafka节点服务器，查看“${KAFKA_HOME}/config”目录下“server.properties”中“listeners”取值获取。例如：10.136.104.56:9092 订阅主题：获取的事件数据来源，即来自于kafka集群的哪个Topic。 例如：__BINGO_PROD_SYS_TOPIC 该章节示例中关闭该开关，使用与AppCube相连的Kafka。

5. 从“图元”中拖拽“投影”图标至左侧画布区域数据源Kafka下方，并在弹出的“添加投影器”页面配置基本信息。单击“保存”。
   图4 “添加投影器”页面
   

   表3 “添加投影器”页面参数说明

   参数名称	参数说明
   标签	该投影的显示标签名。长度不超过64个字节。
   名称	该投影的名称，系统自动生成。
   输入字段	订阅事件的所有事件参数。 请从中选择需要的参数单击，选中的参数会出现在“输出字段”中。
   输出字段	从订阅事件中的事件参数进行筛选后，用到的事件参数。

6. 从“图元”中拖拽“过滤”图标至左侧画布区域投影下方，并在弹出的“添加过滤器”页面配置基本信息和过滤条件。单击“保存”。
   图5 设置过滤条件
   

   表4 过滤条件页面参数说明

   参数名称	参数说明
   标签	该Filter的显示标签名。长度不超过64个字节。
   名称	该Filter的名称，系统自动生成。
   条件设置	过滤条件。 字段：上一步筛选后的事件参数。 比较符：操作符，下拉选择。 值：字段值。 例如：上图中的配置表示提取温度大于60度的事件。 单击“新增行”可以新增条件。

7. （可选）从“图元”页面中拖拽“分组”图标至左侧画布区域过滤器下方，并在弹出的“添加分组器”页面配置基本信息和分组条件。单击“保存”。
   图6 “添加分组器”页面配置
   

   表5 分组条件页面参数说明

   参数名称	参数说明
   标签	该GroupBy的显示标签名。长度不超过64个字节。
   名称	该GroupBy的名称，系统自动生成。
   分组设置	分组条件。 例如上图中配置表示按照相同位置的大于60度的温度取温度平均值。

8. 在“图元”页面右侧拖拽输出源中的Kafka图标至左侧画布区域最下方，设置spark streaming输出事件和输出事件数据的接收方。单击“保存”。
   您可拖拽多个Sink中的Kafka图标，设置多个事件接收方。

   图7 数据接收方配置
   

   表6 数据接收方配置参数说明

   参数名称	参数说明
   标签	数据接收方Kafka的显示标签名。长度不超过64个字节。
   名称	数据接收方Kafka的名称，系统自动生成。
   事件	spark streaming输出的事件。并设置订阅事件和输出事件的参数对应关系。 这里配置为高温报警事件“fire_alarm”。温度监测事件“temperature”的“position”对应高温报警事件“fire_alarm”中的“positionalarm”，温度监测事件的“temperature”对应高温报警事件“fire_alarm”中的“temperaturealarm”。
   发送到外部	是否发送到外部Kafka。 若关闭该开关，表示发送到与AppCube相连的Kafka；若打开该开关，表示发送到外部Kafka，需要配置“Kafka集群地址”和“订阅主题”。 Kafka集群地址：接收事件的Kafka集群地址。 多个节点服务器地址可用“,”间隔，格式为“Kafka节点1的ip:节点1端口号,Kafka节点2的ip:节点2端口号,...”。 获取Kafka的IP和端口号方法：可以通过登录每台Kafka节点服务器，查看“${KAFKA_HOME}/config”目录下“server.properties”中“listeners”取值获取。 例如：10.136.14.56:9092 说明：接收事件的kafka集群不需要一定配置为和AppCube相连。 订阅主题：接收事件数据的kafka Topic。 例如：__BINGO_PROD_SYS_TOPIC 该章节示例中关闭该开关，使用与AppCube相连的Kafka。
   事件属性/属性、值	配置输出事件与接收事件中已定义参数的对应关系。 单击“新增”可添加多行。

9. 单击页面右上方保存按钮。设置该“事件流”基本参数。
   图8 该“事件流”基本参数页面
   

   表7 “事件流”基本参数

   参数名称	参数说明
   标签	该“事件流”的显示标签名。长度不超过64个字节。
   名称	该“事件流”的名称，系统自动生成。
   描述	描述信息。

10. 单击页面上方图标启用“事件流”。