开发常规流数据处理任务¶
常规流数据处理任务需要消费 Kafka Internal Topic 中的全部消息,然后通过 Point Selector 算子过滤出需要使用的数据。流数据处理任务消耗的计算资源较大。
新建常规流数据处理任务¶
前提条件
EnOS 流数据处理服务提供多个版本的算子包,在设计流数据处理任务前,需要先安装对应版本的系统算子包。详细信息,参见 安装算法模板和算子包。
通过以下步骤,使用系统算子包开发常规流数据处理任务:
登录 EnOS 管理控制台,选择 流数据处理 > 流开发,点击任务列表上方的 + 图标。
在 添加流 窗口中,选择 常规 流类型。
选择 新建 方式创建流数据处理任务。也可通过导入配置文件快速创建流数据处理任务。
输入流数据处理任务的名称和描述。
从 模板 下拉菜单中,选择 算子模板。
从 算子版本 下拉菜单中,选择已安装的算子包版本。
在 数据通道 栏中,选择待处理的流数据类型:
- 若数据为接入设备上送的数据,选择 实时通道
- 若数据为通过消息集成模块导入的数据,选择 离线通道
点击 确认,进入数据处理任务设计页面。
设计流数据处理任务¶
通过以下步骤,按业务需求使用算子设计流数据处理任务:
在流数据处理任务设计页面中,点击页面右上角的 Stage Library,从下拉菜单中找到需要使用的算子。点击数据处理算子(如 Point Selector),将其添加到 Pipeline 编辑页面。
拖拽 Stage 和连接线,将添加的 Stage 编排进 Pipeline 中。选中添加的算子,在配置项中完成对该算子的参数配置。
重复步骤1、2,将其他算子编排进 Pipeline 中,并完成各个 Stage 的参数配置。
点击任务栏中的 保存,保存流数据处理任务的配置信息。
完成算子配置后,点击工具栏中的 Validate 图标,检查 Pipeline 和算子参数配置是否正确,并按照检查结果修改配置。
更多使用算子配置 Pipeline 的详细介绍,参考 StreamSets 文档。
导入流数据处理任务配置¶
在数据处理任务设计页面上,除了设计全新的流数据处理任务外,也可通过导入现有流数据处理任务配置文件,快速创建和配置流数据处理任务。