# 如何配置事件流 #### 使用说明 当需要对事件A进行订阅，分析处理后转换成事件B进行发送时，可以通过配置"事件流"，来实现这整个事件流程。目前该复杂场景只应用于开启事件后，spark streaming从kafka中，提取事件并按照定义规则对事件的数据流进行分析处理，然后发送新事件给kafka。为了节省资源，一个租户只能创建一个事件流。 #### 背景信息 首先，您需要理解以下概念： - kafka：是由Apache开发的一个开源流处理平台，是分布式发布、订阅消息系统。kafka主要用于处理活跃的流数据，一般kafka部署为集群模式。 - Topic：用来区分kafka集群中，不同类型消息的主题。例如，应用程序A订阅了Topic1，应用程序B订阅了Topic2而没有订阅Topic1，那么发送到Topic1中的数据将只能被应用程序A读到，而不会被应用程序B读到。 - spark streaming：大规模流式数据处理工具，第三方工具。 #### 场景描述 某仓储公司需要对该公司各个区域设置温度监测，当温度大于60度时，需要触发高温报警。 将温度监测和触发高温报警定义为两个事件，可通过配置"事件流"定义整个事件流程及事件数据筛选规则，开启流程后，spark streaming从kafka中提取温度监测事件，按照筛选规则分析处理后提取温度大于60度的数据，发送高温报警事件给kafka。 图1流程示意图 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885724.png "点击放大") #### 前提条件 - 请联系运维，开启事件流License权限。开启后，才可新建、查看并管理事件流。 - 已创建并启用温度监测事件"temperature"，该事件中自定义参数为文本类型"position"和数字类型"temperature"，分别表示位置和温度，具体操作步骤请参考[如何自定义事件](https://support.huaweicloud.com/usermanual-astrozero/astrozero_05_0082.html)。 - 已创建并启用高温报警事件"fire_alarm"，该事件中自定义参数为文本类型"positionalarm"和数字类型"temperaturealarm"，分别表示高温报警位置和温度，具体操作步骤请参考[如何自定义事件](https://support.huaweicloud.com/usermanual-astrozero/astrozero_05_0082.html)。 #### 操作步骤 1. 参考[登录经典应用设计器](https://support.huaweicloud.com/usermanual-astrozero/astrozero_05_5025.html)中操作，登录经典版应用设计器。 2. 将鼠标放在应用的某个文件夹上（如Logic），单击"+"，选择"事件流"。 开启事件流License权限后，才会有新建事件流的入口。 3. 单击"新建"，进入"事件流"页面。 4. 在"事件流"页面右侧，单击"设置"，进行数据配置。 图2设置页面 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885744.png "点击放大") 窗口长度：事件流处理的数据采集时间间隔，单位为分钟。建议时间间隔不要配置过小，否则系统频繁的建表和执行SQL语句，会影响系统性能。例如：配置为"1"表示spark streaming每隔1分钟去kafka获取事件数据。 5. 在"事件流"页面右侧，单击"图元"，拖拽"输入源"中Kafka图元至左侧画布区域，并配置基本信息，单击"保存"。 输入源中Kafka图标表示数据来自Kafka。 图3数据来源Kafka基本信息页面 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124405.png "点击放大") - 标签：新建数据源Kafka的标签名，用于在界面展示。 - 名称：新建数据源Kafka的名称，系统自动生成。 - 事件：需要订阅的事件。例如，配置为温度监测事件"temperature"。 - 从外部接收：是否从外部Kafka接收事件。 - 关闭该开关，从与AstroZero相连的Kafka接收事件。 - 打开该开关，从外部Kafka接收事件。本示例中关闭该开关，使用与AstroZero相连的Kafka。 - Kafka集群地址：数据源Kafka的集群地址。多个节点服务器地址可用","间隔，格式为"*Kafka* *节点* *1* *的* *ip* :*节点* *1* *端口号* ,*Kafka* *节点* *2* *的* *ip* :*节点* *2* *端口号* ,..."。 获取Kafka的IP和端口号方法：登录每台Kafka节点服务器，查看"${KAFKA_HOME}/config"目录下，"server.properties"中"listeners"的取值。 - 订阅主题：获取的事件数据来源，即来自于kafka集群的哪个Topic。 6. 从"图元"中，拖拽"投影"图元至左侧画布区域数据源Kafka下方，并在弹出的"添加投影器"页面配置基本信息，单击"保存"。 图4"添加投影器"页面 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124441.png "点击放大") - 标签：新建投影的标签名，用于在界面展示。 - 名称：新建投影的名称，系统自动生成。 - 输入字段：订阅事件的所有事件参数。选择所需的参数，单击![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124413.png)，选中的参数会出现在"输出字段"中。 - 输出字段：从订阅事件的事件参数进行筛选后，用到的事件参数。 7. 从"图元"中，拖拽"过滤"图元至左侧画布区域投影下方，并在弹出的"添加过滤器"页面配置基本信息和过滤条件，单击"保存"。 图5设置过滤条件 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124433.png "点击放大") - 标签：新建过滤的标签名，用于在界面展示。 - 名称：新建过滤的名称，系统自动生成。 - 条件设置：设置过滤条件。例如，图中的配置表示提取温度大于60度的事件。 - 字段：上一步筛选后的事件参数。 - 比较符：操作符，直接在下拉框中选择。 - 值：字段值。 8. （可选）从"图元"页面中，拖拽"分组"图元至左侧画布区域过滤器下方，并在弹出的"添加分组器"页面配置基本信息和分组条件，单击"保存"。 图6"添加分组器"页面配置 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001933045128.png "点击放大") - 标签：新建分组的标签名，用于在界面展示。 - 名称：新建分组的名称，系统自动生成。 - 分组设置：设置分组条件。例如，图中配置表示按照相同位置的大于60度的温度取温度平均值。 9. 在"图元"页面右侧，拖拽输出源中的Kafka图元至左侧画布区域最下方，设置spark streaming输出事件和输出事件数据的接收方，单击"保存"。 可以拖拽多个Sink中的Kafka图元，设置多个事件接收方。 图7数据接收方配置 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124417.png "点击放大") - 标签：数据接收方Kafka的标签名，用于在界面展示。 - 名称：数据接收方Kafka的名称，系统自动生成。 - 事件：spark streaming输出的事件。本示例此处配置为高温报警事件"fire_alarm"。温度监测事件"temperature"中的"position"，对应高温报警事件"fire_alarm"中的"positionalarm"。温度监测事件的"temperature"，对应高温报警事件"fire_alarm"中的"temperaturealarm"。 - 发送到外部：是否发送到外部Kafka。若关闭该开关，表示发送到与AstroZero相连的Kafka。若打开该开关，表示发送到外部Kafka，需要配置"Kafka集群地址"和"订阅主题"。本示例中关闭该开关，使用与华为云Astro轻应用相连的Kafka。 - Kafka集群地址：接收事件的Kafka集群地址，例如10.136.14.56:9092。 多个节点服务器地址可用","间隔，格式为"*Kafka* *节点* *1* *的* *ip* :*节点* *1* *端口号* ,*Kafka* *节点* *2* *的* *ip* :*节点* *2* *端口号*,..."。 获取Kafka的IP和端口号方法：登录每台Kafka节点服务器，查看"${KAFKA_HOME}/config"目录下，"server.properties"中"listeners"的取值。 ![](https://support.huaweicloud.com/usermanual-astrozero/public_sys-resources/note_3.0-zh-cn.png) 接收事件的kafka集群，不需要一定配置为和AstroZero相连。 - 订阅主题：接收事件数据的kafka Topic，例如__BINGO_PROD_SYS_TOPIC。 - 事件属性/属性、值：配置输出事件与接收事件中，已定义参数的对应关系。 10. 单击页面右上方保存按钮，设置"事件流"基本参数。 图8该"事件流"基本参数页面 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885728.png "点击放大") - 标签：新建事件流的标签名，用于在界面展示。 - 名称：新建事件流的名称，系统自动生成。 - 描述：新建事件流的描述信息。 11. 单击页面上方的![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124453.png)，启用事件流。 #### 结果验证 1. 创建一个服务编排"fire_alarm"，创建后启用该服务编排，实现当kafka中出现高温报警事件时，AstroZero需要接收该事件，再向第三方系统发送该告警邮件。 1. 鼠标放在应用下的Logic文件夹上，单击"+"，选择"服务编排"。 2. 选择"创建一个新的服务编排"，设置标签和名称为"fire_alarm"，单击"添加"。 3. 在服务编排设计页面左侧，拖拽"逻辑"下的"等待"图元至画布中。 4. 单击![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960284217.png)，在"可编辑的事件"中，单击"新增"，新增"event0"分支。 配置该分支事件为高温报警事件，当系统中出现该事件时，会执行该分支。 图9配置等待图元 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885688.png "点击放大") 5. 拖拽"基本"下的"发送邮件"图元至画布中。 6. 单击![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001933045124.png)，设置发送邮件参数。 配置高温报警邮件的主题和内容，地址填入第三方系统的邮件接收地址。 图10配置发送邮件图元 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885692.png "点击放大") - 直接编辑：手动设置邮件信息。 - 基于模板：基于同一账号或基线中已有的邮件模板，设置邮件主题和内容，方便用户增加效率。 - 模板：选择"基于模板"时，需要选择邮件模板。 - 主题：设置邮件标题。 - 选中"直接编辑"时，需要配置邮件标题。 - 选中"基于模板"时，该参数配置区域置灰。 - 内容：设置邮件内容。 - 选中"直接编辑"时，需要配置邮件内容。 - 选中"基于模板"时，该参数配置区域置灰。 - 自定义变量：选中"基于模板"时，需要配置模板中的参数。在"值"中，直接输入参数取值或者从全局上下文拖拽变量。 - 地址：接收人的邮箱地址。 - 字符串：直接输入分号分隔的字符串或从全局上下文拖拽变量。 - 集合：从全局上下文拖拽集合变量。 - 抄送：抄送人员的邮箱地址，直接输入分号分隔的字符串或者从全局上下文拖拽变量。 - 密送：密送发送人员的邮箱地址，直接输入分号分隔的字符串或者从全局上下文拖拽变量。 7. 按照下图连接所有图元，等待图元和发送邮件图元之间选择"event0"分支。 图11连接所有图元 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001933045144.png "点击放大") 8. 单击页面上方的![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885736.png)，保存服务编排。 9. 保存成功后，单击![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885700.png)，启用服务编排。 2. 创建一个服务编排"send_fire_event"，将温度监测事件"temperature"中位置参数"position"赋值为仓储公司某区域，温度"temperature"赋值为70度，启用该Flow，用于模拟事件发送。 1. 鼠标放在应用下的Logic文件夹上，单击"+"，选择"服务编排"。 2. 选择"创建一个新的服务编排"，设置标签和名称为"send_fire_event"，单击"添加"。 3. 在服务编排编辑器页面左侧，拖拽"基本"下的"发送事件"图元至画布中。 4. 单击![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960284265.png)，设置发送事件参数。 图12配置发送事件图元 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124461.png "点击放大") - 事件：选择待发送的事件。 - 分区字段：从事件中选择一个自定义参数用作Kafka的分区字段，则根据该字段值进行路由，相同的值将路由到同一个Kafka分区。如果不指定，则默认随机路由到不同Kafka分区，从而提升并发处理性能。 - 发送到外部：系统内部有配置与AstroZero相连Kafka的固定Topic"__BINGO_SYS_TOPIC"。若不勾选"发送到外部"，表示将事件数据发到默认的Topic上。若勾选"发送到外部"，则表示将事件数据发送到与AstroZero相连Kafka的其他Topic上。 - 延迟到事务结束（提交或回滚）后才发送：是否延迟到该服务编排事务结束后，才发送事件。 - 主题：勾选"发送到外部"时，才显示该参数，表示非系统配置的默认Topic。 - 目标/源：配置事件数据。 本示例中，"目标"请从下拉框中选择事件的自定义参数"temperature"和"position"，在"源"中给目标赋值。 5. 按照下图连接所有图元。 图13连接图元 ![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001933045148.png "点击放大") 6. 单击页面上方的![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124429.png)，保存服务编排。 7. 保存成功后，单击![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001933045092.png)，启用服务编排。 3. 在"事件流"列表页面，单击该事件流程所在行的![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001960124437.png)，运行该事件流程。 4. 单击服务编排"send_fire_event"编辑器上方的![](https://support.huaweicloud.com/usermanual-astrozero/zh-cn_image_0000001932885704.png)，执行服务编排。 不用输入任何输入参数，直接单击"运行"。 5. 参考上一步，执行服务编排"fire_alarm"。 不用输入任何输入参数，单击"运行"。 6. 打开第三方系统邮箱，成功收到高温警报邮件，验证成功。