通过规则引擎的云产品流转功能，物联网平台可将指定Topic的数据，流转至其他的Topic和阿里云产品中。本文将为您介绍设置数据流转规则的完整操作步骤，依次是创建规则、编写处理数据的SQL、设置数据流转目的地、设置流转失败的数据转发目的地。

操作步骤

  1. 登录物联网平台控制台。
  2. 在实例概览页面，找到对应的实例，单击实例进入实例详情页面。
    注意 在中国地域，目前仅华东2（上海）地域开通了公共实例服务。
    实例概览
  3. 在左侧导航栏，选择规则引擎 > 云产品流转。
  4. 在云产品流转页面，单击创建规则。
    注意 若当前页面显示新版功能，先单击右上角返回旧版，进入旧版功能页面，再单击创建规则。
  5. 填写参数后，单击确认。
    参数 描述
    规则名称 输入规则名称。支持中文、英文字母、日文、数字、下划线（_）和短划线（-），长度为1~30个字符，一个中文或日文占2个字符。
    数据格式 选择该规则处理数据的格式。可选：JSON和二进制。
    说明
    • 因数据流转基于Topic处理数据，此处的数据格式需与被处理Topic中的数据格式保持一致。
    • 若选择为二进制，该规则不能处理基础通信Topic、物模型通信Topic的消息，且不能将数据转发至实例内的时序数据存储、时序数据库、表格存储和云数据库RDS版。
    规则描述 规则描述信息。
  6. 规则创建成功后，将跳转到数据流转规则页面。您需编辑处理消息数据的SQL、设置数据转发目的地、流转失败数据转发目的地。
    1. 单击编写SQL，编写处理消息字段的SQL。

      SQL编写方法，可参见SQL表达式和函数列表。

      参数 描述
      规则查询语句 系统会在这里，根据您设置的字段、Topic和条件自动补充完整规则查询语句。
      字段

      指定要处理的消息内容字段，即SQL中SELECT后的内容。

      例如，填入deviceName() as deviceName，则表示需筛选出消息中的deviceName字段内容。字段中可使用的函数，请参见函数列表。

      说明 基础通信Topic、物模型通信Topic中的数据为Alink JSON格式，流转到规则引擎前，会经过物模型解析，具体说明，请参见数据流转过程。解析后的数据格式，请参见数据格式。编写SQL字段时，需按照解析后的数据格式来编写。
      Topic

      选择需要处理的消息Topic，即SQL中FROM后的内容。可选Topic，请参见下表《Topic说明》。

      注意 当规则的数据格式是二进制时，仅支持选择自定义。
      条件 设置规则触发条件，即SQL中WHERE后的内容。
      表 1. Topic说明
      Topic 说明 相关文档
      自定义 流转自定义数据格式消息的Topic，与自定义Topic的格式相同。格式为：/${productKey}/${deviceName}/user/${TopicShortName}。

      其中${TopicShortName}为自定义的Topic类，即自定义Topic的后缀。

      支持使用通配符（+）和（#）：

      • 全部设备（+）：指定产品下所有设备。
      • /user/#：指定设备的所有自定义Topic。
      		自定义Topic
      设备状态变化通知 流转设备上下线状态变更消息的Topic：/as/mqtt/status/${productKey}/${deviceName}。 设备上下线状态
      物模型数据上报 包含：
      • 流转设备上报属性数据的Topic：/${productKey}/${deviceName}/thing/event/property/post。
      • 流转设备上报事件数据的Topic：/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post。
      • 流转设备批量上报属性数据的Topic：/${productKey}/${deviceName}/thing/property/batch/post。
      • 流转设备批量上报事件数据的Topic：/${productKey}/${deviceName}/thing/event/batch/post。
      • 流转设备响应云端命令返回消息的Topic：/${productKey}/${deviceName}/thing/downlink/reply/message。
      对应设备Topic如下：
      • 设备上报属性的Topic：/sys/${productKey}/${deviceName}/thing/event/property/post。
      • 设备上报事件的Topic：/sys/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/post、/sys/${productKey}/${deviceName}/thing/event/${tsl.functionBlockId}:{tsl.event.identifier}/post。
      • 设备批量上报属性、事件数据的Topic：/sys/${productKey}/${deviceName}/thing/event/property/batch/post。
      设备生命周期变更 流转设备创建、删除、禁用、启用等消息的Topic：/${productKey}/${deviceName}/thing/lifecycle。 设备生命周期变更
      网关发现子设备上报 网关设备特有的Topic：/${productKey}/${deviceName}/thing/list/found，将发现的子设备信息上报给物联网平台，然后进行流转。 网关发现子设备
      设备拓扑关系变更 网关设备特有Topic：/${productKey}/${deviceName}/thing/topo/lifecycle，流转子设备和网关之间的拓扑关系建立和解除消息的Topic。 设备拓扑关系变更
      对应设备上报数据的Topic：/sys/${productKey}/${deviceName}/thing/topo/change。 通知网关拓扑关系变化
      设备标签变更 流转设备标签信息变更的Topic：/${productKey}/${deviceName}/thing/deviceinfo/update。 设备标签变更
      对应设备上报数据的Topic：/sys/${productKey}/${deviceName}/thing/deviceinfo/update。 标签信息上报
      物模型历史数据上报 包含：
      • 流转设备上报历史属性数据的Topic：/${productKey}/${deviceName}/thing/event/property/history/post。
      • 流转设备上报历史事件数据的Topic：/${productKey}/${deviceName}/thing/event/${tsl.event.identifier}/history/post。
      对应设备上报物模型历史数据的Topic：/sys/${productKey}/${deviceName}/thing/event/property/history/post。 物模型历史数据上报
      OTA升级设备状态通知 包含：
      • 流转设备上报OTA升级结果的Topic：/${productKey}/${deviceName}/ota/upgrade。
      • 流转设备上报OTA升级进度的Topic：/${productKey}/${deviceName}/ota/progress/post。
      对应设备上报升级进度的Topic：/ota/device/progress/${productKey}/${deviceName}。 设备上报升级进度
      OTA模块版本号上报 流转设备上报OTA模块版本号变更的Topic：/${productKey}/${deviceName}/ota/version/post。 OTA模块版本号变更通知
      对应设备上报OTA模块版本的Topic：/ota/device/inform/${productKey}/${deviceName}。 设备上报OTA模块版本
      OTA升级批次状态通知 物联网平台通知OTA升级批次状态变化的Topic：/${productKey}/${packageId}/${jobId}/ota/job/status。 OTA升级批次状态通知
      任务事件 包含：
      • 流转设备任务状态通知的Topic：/sys/uid/${uid}/job/${jobId}/lifecycle。
      • 流转实例迁移任务状态通知的Topic：/sys/uid/${uid}/distribution/${jobId}/lifecycle。
        说明 迁移产品的名称为实例迁移的任务名称。
      孪生节点属性变更 流转数字孪生节点属性数据的Topic：/sys/uid/${uid}/digitaltwin/${dtInstanceId}/${nodeId}/property/update。 数字孪生节点属性变更
    2. 单击转发数据一栏的添加操作，设置数据转发目的地。数据转发配置的具体示例，请参见数据流转使用示例目录下的具体文档。
      说明 最多可为一个规则创建10个数据转发操作。
      数据转发时，因选择的数据目的地（云产品）出现异常情况导致转发失败时：
      • 转发到RocketMQ、RDS、TSDB等类型的云产品时，若云产品资源变化导致云产品无法访问，系统将停止执行转发，并在规则列表中显示状态为规则异常。您需要重新设置数据转发目的地。
      • 其他异常情况，系统将间隔1秒、3秒、10秒进行3次重试（重试策略可能会调整）。3次重试均失败后，消息会被丢弃。如果您对消息可靠性要求比较高，可以进行：添加错误操作，将重试失败的消息，转发到其他云产品中。
    3. 单击转发错误操作数据一栏的添加错误操作，设置参数，将重试失败的错误消息转发至指定位置。
      注意
      • 最多支持添加一个错误操作。
      • 正常操作和错误操作的转发目的地不能是相同的云产品。例如，不能同时转发到表格存储。
      • 错误消息转发失败后，不会再进行重试。
      • 这里的错误消息仅针对因其他云产品实例问题导致的规则引擎转发失败错误。
      消息转发至云产品失败后，会进行重试。若重试失败，将根据错误操作数据转发的设置转发错误消息。

      错误消息格式： 

      {
      "ruleName":"",
      "topic":"",
      "productKey":"",
      "deviceName":"",
      "messageId":"",
      "base64OriginalPayload":"",
      "failures":[
        {
          "actionType":"OTS",
          "actionRegion":"cn-shanghai",
          "actionResource":"table1",
          "errorMessage":""
        },
        {
          "actionType":"RDS",
          "actionRegion":"cn-shanghai",
          "actionResource":"instance1/table1",
          "errorMessage":""
        }
      ]
}

      错误消息参数说明如下：

      参数 说明
      ruleName 规则名称。
      topic 消息来源Topic。
      productKey 产品ProductKey。
      deviceName 设备名称。
      messageId 云端消息ID。
      base64OriginalPayload Base64编码后的原始数据。
      failures 错误详情。可能会有多个。
      actionType 出错操作的类型。
      actionRegion 出错操作的地域。
      actionResource 出错操作的目的资源。
      errorMessage 错误信息。
  7. 所有设置完成后，返回至云产品流转页面，单击规则对应的启动。规则启动后，数据即可按照规则进行转发。
    您也可以单击：
    • 查看：在数据流转规则页面，修改规则的具体设置。
    • 删除：删除对应规则。
      说明 运行中的规则不可删除。
    • 停止：停止对应规则转发数据。