工业互联网网络运维 课件 任务6.1-3 IoT平台数据管理-IoT平台数据流管理

主讲：胡春芬工业互联网网络运维

平台数据流主要包含新建/导出数据流、数据流列表展示、数据流定义/启停三项功能。通过拖拉拽的可视化配置工具，构建数据处理流程引擎，实现数据的采集、加工、转发、存储等功能。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

平台数据流提供布局环境以及丰富的组件库，组件库包括4种类型：Origins、Processors、Destinations、Executors。Origins：数据源组件

对接IOT网关等物联网设备。配置数据源的相关属性，完成数据的同步接收。支持文件数据源、http数据源、OPCUA、Restful、Udp、WebSocket等数据源，实现了数据接口的广泛适配性。注意：一个工作流只能存在一个origin组件。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理Processors：数据处理组件

支持消息的订阅、发布，以及在海量设备下的高并发响应。支持脚本语言的二次开发，数据流中的数据节点在线清洗、加工，再次数据打包传输。数据处理组件丰富，包含：数据转码处理、数据解析器、数据延迟处理、数据特征、正则表达式、数据孪生、数据扁平化处理等组件。可以根据不同的业务场景，进行数据处理组件的逻辑组合，搭建新的业务流程引擎。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理Destinations：目标阶段

表示管道的目标，可以在管道中使用一个或多个目标，可以根据管道的执行模式使用不同的目的地，包括：Trash

：垃圾桶组件，用于将数据丢弃HTTPClient：HTTP客户端，将数据写入到支持HTTP协议的数据服务端WebSocketClient：WebSocket客户端，通过该组件将数据发送到WebSocket服务端任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理Destinations：目标阶段MQTTPublisher：将数据输出到MQTT消息队列中Redis：将数据输出到RedisSendResponsetoOrigin：用在微服务工作流中，将数据返回到用户端ToError//将数据流发送到管道错误处任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理Executors：Executor类组件主要在收到事件时会触发任务。包括：Email:在收到事件后向配置的收件人发送自定义电子邮件JDBCQuery:收到事件记录后运行用户定义的SQL查MapReduce:收到事件记录后启动指定的MapReduce作业Shell:在接收到事件记录执行shell脚本PipelineFinisher:收到事件记录后停止并将管道转换为Finished状态任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

以数据采集—数据处理—数据转发—数据销毁/转发流程为例，系统平台常用的组件介绍如下。“MQTTSubscriber”组件

“MQTTSubscriber”组件为源数据组件，负责实现与现场IoT硬件网关设备的数据对接。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

首先点击界面右上角的“状态库”按钮，会出现各类组件的下拉列表。下拉框中选择“Origins”，然后在下方搜索栏中输入：MQTT，在出现的列表清单中，点击“MQTTSubscriber”组件，该组件就出现在左侧的编辑区域中。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

选中“MQTTSubscriber1”的组件下方区域显示点的详细属性信息，可进行填写及编辑操作。分类属性信息配置详情MQTT服务器鉴权信息BrokerURLtcp://:9993ClientIDXXXXXX（IoT平台提供）UseCredentials勾选UsernameXXXXXX（IoT平台提供）PasswordXXXXXXXX（IoT平台提供）数据信息TopicFilter为TOPIC订阅主题，订阅主题的格式为：CompayID/IoT网关数据转发名称/updateDataFormat下拉列表中选择JSON任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理“MQTTSubscriber1”配置界面任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理“JythonEvaluator”组件

“JythonEvaluator”组件为数据处理组件，用以对接收到的数据进行数据运算处理。如：对部分数据做除10处理。通过此组件处理后的数据继续输出到后续组件节点。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

首先点击界面右上角的“状态库”按钮，会出现各类组件的下拉列表。下拉框中选择“Processors”，然后在下方搜索栏中输入：jython，在出现的列表清单中，点击“JythonEvaluator”组件，该组件就出现在左侧的编辑区域中。

选中“JythonEvaluator1”组件，点击配置——Jython，进行数据处理脚本的编写工作，主要的编写区域为“Script”区域，这儿为关键的数据处理脚本。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理“Script”脚本区域编写代码示例：importjson,timeimportsysreload(sys)sys.setdefaultencoding('utf-8')debug=Falseforrecordinrecords:try:mydata1=record.value

resultData={}dealTime=int((time.time())*1000)resultData["profileVersion"]=1resultData["timestamp"]=dealTimeresultData["reported"]=mydata1["reported"]resultData["deviceId"]="plc"resultData["type"]=""resultData["dtName"]="运维平台测试".decode("utf-8")record.value=resultDataoutput.write(record)

exceptExceptionase:

error.write(record,str(e))其中resultData["dtName"]="运维平台测试".decode("utf-8")里的"dtName"需要配置成和新建设备时所取名称一致，数据流才指向该设备，并做数据显示。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理“DigitalTwinHaierProcessor”组件“DigitalTwinHaierProcessor”组件为数据处理组件，用以对接收到的数据进行分发，转到到“设备管理”功能模块，进行设备实时数据的展示，形成数据应用的闭环。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

首先点击界面右上角的“状态库”按钮，会出现各类组件的下拉列表。下拉框中选择“Processors”，然后在下方搜索栏中输入：Digital，在出现的列表清单中，点击“DigitalTwinHaierProcessor”组件，该组件就出现在左侧的编辑区域中。

选中“DigitalTwinHaierProcessor1”组件，点击配置——DigitalTwinHaier，主要配置“DigitalTwinAPIURL”，此处输入目标对象APIURL地址，实现数据的网络传递。此处配置为:20011，URL地址和端口号由平台统一下发。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理“Trash”组件

如果在数据处理完毕之后，无后续数据推送或者再次处理的步骤，可以在流程最后执行销毁操作。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

在“状态库”下拉框中选择“Destinations”，然后在下方搜索栏中输入：Trash，从右侧展示列表中点击“Trash”组件，该组件就出现在左侧的编辑区域中。

该组件为数据销毁不做处理，无设置内容。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理“HTTPClient”组件

如果数据在处理完毕后，数据需推送至平台时序数据库存储，则可在“状态库”下拉框中选择“Destinations”，然后在下方搜索栏中输入：http，从右侧展示列表中点击“HTTPClient”组件，该组件就出现在左侧区域。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

选中“HTTPClient1”组件，点击配置——HTTP，主要配置“ResourceURL”，此处输入目标对象ResourceURL地址，实现数据的网络传递。此处配置为:20010/api/put，URL地址和端口号由平台统一下发。点击配置——DataFormat，从下拉列表中选择Text格式。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

流程引擎中各个组件之间通过箭头流程关联，箭头流程提供了众多的处理器功能供选择调用。各组件功能配置完成后，使用箭头将各组件进行连接，以关联流程。连接过程鼠标左键点击组个的旁边的小圆圈，并拖动至另一个组件的小圆圈，可以看见箭头连接完成。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

全部组件配置好后，可以看到仍有报错信息，选中报错指示符号“叹号”，点击配置——ErrorRecords，修改ErrorRecords，从下拉列表中选择Discard(Library:Basic)，修改过程如下图所示。修改完成后报错指示符号消失。任务1

IoT平台数据管理任务1

IoT平台数据管理任务实现3——IoT平台数据流管理

全部组件配置好后，点击右上角“启动”来使整个数据流运行，数据流运行后，可将MQTT服务器数据传送至设备管理。任务1

IoT平台数据管理任务1

IoT平台数据管理任