本文将以一个简单的案例,为您介绍如何使用Stream Studio进行实时计算(原流计算)任务的开发和管理,完成一个实时计算任务的新建、开发、发布、启动、停止和下线等所有操作。

前提条件

在您开始快速入门前,请首先绑定实时计算项目,并至少开通DataWorks标准版。

背景信息

说明 仅独享模式支持资源引用固定列切分动态列切分行切分等功能,本文以独享模式的开发为例进行操作。
  • 数据源:1个DataHub表(Topic),包括m_name,id,m_type,tag4个字段。数据源
    说明 您需要首先创建好Datahub topic,创建方法请参见Datahub Web控制台介绍
  • 数据处理:针对tag字段进行分割,分割符为;,分割后成为colormodeweight3个字段。
  • 数据结果输出:最后选择id,m_type,weight写入至日志服务(SLS)表中。
    说明 您需要首先创建好日志服务项目与Logstore,创建方法请参见操作Logstore

新建业务流程

  1. 进入Stream Studio页面后,单击新建业务流程
    新建业务流程
    您也可以鼠标悬停至新建图标,单击业务流程业务流程
  2. 新建业务流程对话框中,填写业务名称描述
  3. 单击新建

新建实时计算任务

  1. 右键单击新建的业务流程,选择新建任务
    新建任务
  2. 新建节点对话框中,输入节点名称,并选择目标文件夹
  3. 单击提交,即可进入组件页面。
    组件
  4. 单击左侧导航栏中的资源引用,勾选PUBLIC_COMMON
    资源引用
    如果不勾选,后续使用固定列切分组件时会收到如下提示:不勾选
    说明 仅独享模式的工作空间会显示资源引用界面。

编辑实时计算任务

  1. 组件页面新增数据源。
    1. 拖拽数据源表 > DataHub至右侧面板。编辑实时计算
    2. 单击DataHub组件,填写页面右侧的参数配置
      填写参数
      配置 说明
      血缘表名(任务唯一) 对应建表语句中的表名,全局唯一,如果不填将使用真实表名填充。
      定义列 定义输出的列值。
      单击自定义,单击选择字段对话框中的添加,输入名称和类型,单击确认,即可添加输出列。定义列
      连接地址 对应SQL中with参数endPoint
      accessId 对应SQL中with参数accessId
      accessKey 对应SQL中with参数accessKey
      项目名 对应SQL中with参数project
      主题 对应SQL中with参数topic
      开始时间 对应SQL中with参数startTime
      读取最大重试次数 对应SQL中with参数maxRetryTimes,默认值为20
      读取重试间隔 对应SQL中with参数retryIntervalMs,单位为毫秒,默认值为1,000
      单次读取条数 对应SQL中with参数batchReadSize,默认值为10
      单行字段条数检查策略 对应SQL中with参数lengthCheck,默认值为NONE
      调试开关 对应SQL中with参数columnErrorDebug,是否打开调试开关。如果打开,会打印解析异常的日志,您可以在运维中心查看任务详情。
  2. 数据处理。
    1. 拖拽一个固定列切分组件至面板中,进行字段切分。
      固定列切分
      说明 仅独享模式的工作空间支持固定列切分组件。
    2. 连线DataHub组件至固定列切分组件。
      固定列切分组件
    3. 单击DataHub组件,在页面右侧的参数配置选择字段tag
      tag
    4. 修改列分隔符为(;)。
      修改列分隔符
    5. 单击自定义,在选择字段对话框中,输入切分后字段的位置名称,单击确认
      确认
    6. 选择需要输出的字段:拖拽出Select组件。select
    7. 连接固定列切分组件与Select组件。单击Select节点,在面板中单击已选择0字段联线
    8. 选择字段对话框中,勾选idm_typeweight字段,单击确认确认
  3. 输出数据结果(以输出上述处理结果至LogService为例)。
    1. 组件面板中,拖拽LogService组件至右侧面板。logservice
    2. 连接Select组件和LogService组件。连线
    3. 单击LogService组件,填写组件参数。
      配置 说明
      选择输出字段 单击已选择0字段,选择所有字段单击确认
      连接地址 https://streamstudio-shanghai-test.cn-shanghai.log.aliyuncs.com
      项目名 填写相应的项目名称。
      accessId 填写LogService账号的AccessID,在本文中使用的是同一个AccessID。
      accessKey 填写LogService账号的AccessKey,在本文中使用的是同一个AccessKey。
      日志名 填写相应的logStore名称。
  4. 切换DAG模式与SQL模式。

    Stream Studio支持DAG模式编辑和SQL模式编辑流任务,两种模式对等,且支持互相转换。

    1. 单击切换为SQL模式,即可将DAG节点转换为对应的SQL。切换SQL
    2. SQL视图如下所示,单击切换为DAG模式即可再次切换回DAG模式。切换DAG
  5. 设置执行计划。
    1. 单击右侧导航栏中的执行计划执行计划
    2. 请注意对话框中,单击确认确认
    3. 单击保存执行计划执行计划

发布实时计算任务

编辑好流任务之后即可进行任务发布。

  1. 单击保存,然后单击发布(发布时会检查是否保存,如果没有保存也会提示保存)。发布
  2. 发布任务对话框中,输入版本备注,单击确定发布任务
  3. 发布成功后,单击右上角的运维,即可进入运维页面查看任务的状态并进行管理。

运维实时计算任务

进入运维页面后,您可以启动下线监控相应任务,也可以批量启动暂停恢复停止任务。运维
  • 启动任务。
    1. 单击相应任务后的启动
    2. 设置启动点位对话框中,选择启动点位启动点位
    3. 任务启动后,单击任务名称即可查看是否运行成功及其详细信息。详细信息
  • 停止和下线任务。

    单击相应任务后的停止,即可停止该任务的运行。

    任务停止运行后,单击下线,即可下线已停止运行的任务。

执行结果

至此便完成一个实时计算任务的新建、开发、发布、启动、停止和下线等操作。