开发常规流数据处理任务¶

常规流数据处理任务需要消费 Kafka Internal Topic 中的全部消息，然后通过 Point Selector 算子过滤出需要使用的数据。流数据处理任务消耗的计算资源较大。

新建常规流数据处理任务¶

前提条件

EnOS 流数据处理服务提供多个版本的算子包，在设计流数据处理任务前，需要先安装对应版本的系统算子包。详细信息，参见安装算法模板和算子包。

通过以下步骤，使用系统算子包开发常规流数据处理任务：

登录 EnOS 管理控制台，选择 流数据处理 > 流开发，点击任务列表上方的 + 图标。
在 添加流 窗口中，选择常规流类型。
选择新建方式创建流数据处理任务。也可通过导入配置文件快速创建流数据处理任务。
输入流数据处理任务的名称和描述。
从模板下拉菜单中，选择 算子模板。
从 算子版本 下拉菜单中，选择已安装的算子包版本。
在 数据通道 栏中，选择待处理的流数据类型：
- 若数据为接入设备上送的数据，选择 实时通道
- 若数据为通过消息集成模块导入的数据，选择 离线通道
点击确认，进入数据处理任务设计页面。

通过以下步骤，按业务需求使用算子设计流数据处理任务：

在流数据处理任务设计页面中，点击页面右上角的 Stage Library，从下拉菜单中找到需要使用的算子。点击数据处理算子（如 Point Selector），将其添加到 Pipeline 编辑页面。
拖拽 Stage 和连接线，将添加的 Stage 编排进 Pipeline 中。选中添加的算子，在配置项中完成对该算子的参数配置。
重复步骤1、2，将其他算子编排进 Pipeline 中，并完成各个 Stage 的参数配置。
点击任务栏中的保存，保存流数据处理任务的配置信息。
完成算子配置后，点击工具栏中的 Validate 图标，检查 Pipeline 和算子参数配置是否正确，并按照检查结果修改配置。