基于云边协同的物联网数据实时处理方法及系统与流程

本技术涉及物联网实时数据处理，具体涉及一种基于云边协同的物联网数据实时处理方法及系统。

背景技术：

1、随着物联网设备数量及数据采集频率不断增加，物联网系统需要持续接收并处理大量设备数据。现有物联网系统通常通过数据网关接收设备数据，并将接收的数据发送至云端进行集中处理。当短时间内接收的数据量超过云端处理能力时，容易发生数据积压，造成数据处理时延增加，影响设备状态的及时更新。

2、在云端处理过程中，数据处理任务的数量和等待时间不断变化，云端计算节点的运行负载也会随之变化。采用固定的数据写入速率和任务处理方式，难以适应上述变化，并可能在处理规则更新或处理阶段执行异常时造成处理停滞或阶段衔接异常。因此，如何在数据量和云端处理状态发生变化时保持物联网实时数据处理过程的连续性，是亟待解决的问题。

技术实现思路

1、本技术提供了一种基于云边协同的物联网数据实时处理方法及系统，至少用于解决现有技术存在的问题。

2、本技术实施例的第一方面，提供了一种基于云边协同的物联网数据实时处理方法，包括以下步骤：

3、s1、通过数据接入网关接收物联网设备原始数据，按照设备模型生成包含设备标识、采集时间和测点数据的标准化数据，并写入云端流式数据通道；s2、按采集时间将标准化数据划分为时间窗口，生成带任务标识、任务优先级和规则版本号的数据处理任务，将数据处理任务划分为具有执行依赖关系的多个处理阶段，各处理阶段携带任务标识；s3、读取云端当前规则版本号，版本号改变时，仅更新首个处理阶段未开始执行的数据处理任务的规则版本号，并保持首个处理阶段已开始执行的数据处理任务的规则版本号以及云端计算节点加载的原规则版本号，直至已开始执行任务完成；s4、监测云端流式数据通道的缓存队列长度，根据预设速率映射关系确定目标写入速率并生成控制指令，目标写入速率随缓存队列长度增加而降低或保持不变；s5、计算数据处理任务的等待时长，读取云端计算节点的节点负载值和已加载规则版本号，根据任务优先级和等待时长生成任务处理顺序，并从已加载规则版本号与数据处理任务的规则版本号相同的云端计算节点中，按节点负载值由低至高确定各处理阶段的目标计算节点；s6、发送控制指令至数据接入网关以调节数据写入速率，按任务处理顺序，将首个处理阶段和前序处理阶段已完成的后续处理阶段发送至目标计算节点执行，携带同一任务标识的全部处理阶段完成后，合并阶段处理结果并按设备标识更新物联网设备的云端状态。

4、通过将设备原始数据转换为具有统一字段的数据，并按照采集时间形成具有执行依赖关系的数据处理任务，可以使不同处理阶段围绕同一任务有序执行。根据缓存队列长度调整数据写入速率，可以使数据写入速度随队列积压程度变化，有助于减缓缓存队列继续增长。结合任务优先级和等待时长确定任务处理顺序，再根据规则版本匹配关系和节点负载值确定目标计算节点，可以兼顾任务等待情况、规则适配情况和节点任务负担。在规则版本变化时保持已开始执行任务使用的规则版本，并在全部处理阶段完成后合并处理结果，可以降低规则切换或者阶段执行次序异常对任务处理过程的影响。因此，在数据量和云端处理状态发生变化时，有助于提高物联网实时数据处理过程的连续性。

5、进一步地，在步骤s3中，将与云端当前规则版本号不同的规则版本号确定为非当前规则版本号；针对各非当前规则版本号，分别统计规则版本号为对应非当前规则版本号、首个处理阶段已开始执行且尚未完成的数据处理任务的任务数量；当任一非当前规则版本号对应的任务数量大于零时，禁止已加载对应非当前规则版本号的云端计算节点卸载对应非当前规则版本号；当任务数量减小至零时，允许云端计算节点卸载对应非当前规则版本号。

6、通过按照非当前规则版本号统计尚未完成的任务，可以依据各版本的实际使用情况确定卸载时机，使任务执行期间能够继续调用对应规则版本，并在相关任务完成后释放相应节点资源。

7、进一步地，在步骤s4中，将缓存队列长度的取值范围划分为多个长度区间，并为各长度区间设置对应的写入速率，其中，长度区间对应的缓存队列长度越大，对应的写入速率越小或保持不变；确定当前缓存队列长度所属的长度区间，将所属长度区间对应的写入速率确定为目标写入速率。

8、通过设置缓存队列长度区间与写入速率之间的对应关系，可以直接根据当前缓存队列长度确定目标写入速率，并在缓存队列长度增大时限制写入速率的增长，有助于减缓缓存队列继续积压。

9、进一步地，在步骤s5中，针对每一云端计算节点，统计已分配至对应云端计算节点且尚未开始执行的处理阶段数量以及正在执行的处理阶段数量；将尚未开始执行的处理阶段数量乘以第一预设权重，得到第一负载分量，将正在执行的处理阶段数量乘以第二预设权重，得到第二负载分量；将第一负载分量与第二负载分量相加，得到对应云端计算节点的节点负载值；在处理阶段被分配、开始执行、执行完成或执行失败时，更新执行对应处理阶段的云端计算节点的节点负载值；在处理阶段被重新分配时，分别更新原目标计算节点和新目标计算节点的节点负载值。

10、通过对尚未开始执行和正在执行的处理阶段数量分别进行加权，可以同时反映节点的待执行任务量和正在执行任务量。随着处理阶段状态及分配关系变化更新节点负载值，可以使节点负载值与相应节点当前承担的处理阶段相对应。

11、进一步地，在步骤s5中，读取各云端计算节点的运行状态、节点标识、节点负载值和已加载规则版本号；筛选运行状态为可用状态且已加载规则版本号与数据处理任务的规则版本号相同的云端计算节点；按照节点负载值由低至高、正在执行的处理阶段数量由少至多以及节点标识的预设顺序，依次排列筛选得到的云端计算节点，并将排列在首位的云端计算节点确定为目标计算节点。

12、通过先筛选运行状态可用且规则版本匹配的云端计算节点，再按照节点负载值、正在执行的处理阶段数量和节点标识依次排序，可以减少处理阶段被分配至不可用节点或版本不匹配节点的情况，并在多个候选节点之间获得确定的选择结果。

13、进一步地，在步骤s2中，读取数据处理任务包含的各测点数据对应的预设实时等级，将其中最高的预设实时等级确定为数据处理任务的任务实时等级，并根据任务实时等级确定初始任务优先级；生成任务处理顺序时，按照任务优先级由高至低排列数据处理任务，并按照等待时长由长至短排列任务优先级相同的数据处理任务；当数据处理任务的等待时长达到预设等待阈值且任务优先级低于预设最高优先级时，将任务优先级提高一个等级，并重新生成任务处理顺序。

14、通过根据测点数据的实时等级确定初始任务优先级，并结合任务等待时长调整任务处理顺序，可以优先安排实时等级较高的任务，同时使等待时长达到阈值的任务获得提前处理的机会。

15、进一步地，在步骤s2中，为多个处理阶段设置阶段标识；在步骤s6中，使各阶段处理结果携带对应处理阶段的任务标识和阶段标识，并在将处理阶段发送至目标计算节点执行后，检测处理阶段的执行状态；当处理阶段执行失败或达到预设执行时限仍未返回阶段处理结果时，从除执行处理阶段的目标计算节点以外的云端计算节点中，筛选运行状态为可用状态且已加载规则版本号与处理阶段所属数据处理任务的规则版本号相同的云端计算节点，得到候选计算节点；当存在候选计算节点时，按照节点负载值由低至高、正在执行的处理阶段数量由少至多以及节点标识的预设顺序排列候选计算节点，将排列在首位的候选计算节点确定为新的目标计算节点，并将处理阶段发送至新的目标计算节点重新执行；对于携带相同任务标识和阶段标识的阶段处理结果，保留首先接收的执行成功的阶段处理结果，并舍弃随后接收的阶段处理结果。

16、通过在处理阶段执行失败或超时时将处理阶段发送至其他规则版本匹配的可用节点，可以为未完成的处理阶段提供重新执行路径。根据任务标识和阶段标识识别同一处理阶段的多个返回结果，可以减少重新执行结果与迟到结果被重复合并的情况。

17、进一步地，在步骤s2中，按照设备标识对标准化数据进行分组；将同一分组内采集时间位于同一时间窗口的标准化数据划入同一数据处理任务；根据设备标识、时间窗口的起始时间和结束时间生成任务标识。

18、通过设备标识和时间窗口归集标准化数据，并利用设备标识及时间窗口的起止时间生成任务标识，可以区分不同设备或不同时间范围形成的数据处理任务，为处理阶段及阶段处理结果关联相应任务提供标识依据。

19、进一步地，在步骤s1中，读取设备模型记载的字段映射关系和数据类型转换规则；根据字段映射关系，确定物联网设备原始数据中的设备标识字段、采集时间字段和测点数据字段；从设备标识字段、采集时间字段和测点数据字段中分别提取设备标识、采集时间和测点数据，根据数据类型转换规则转换测点数据的数据类型，并将设备标识、采集时间和转换后的测点数据组合为标准化数据。

20、通过字段映射关系提取设备标识、采集时间和测点数据，并按照数据类型转换规则转换测点数据，可以将字段设置或数据类型不同的设备原始数据转换为统一的数据结构，以供后续时间窗口划分和任务生成使用。

21、进一步地，在步骤s2中，为多个处理阶段设置阶段标识和处理层级，将不具有前序处理阶段的处理阶段的处理层级值设置为一，并将各后续处理阶段的处理层级值设置为对应直接前序处理阶段中的最大处理层级值加一；在步骤s6中，将各处理阶段携带的任务标识、阶段标识和处理层级值写入对应的阶段处理结果，按照处理层级值由小至大排列携带相同任务标识的阶段处理结果，按照阶段标识的预设顺序排列处理层级值相同的阶段处理结果，并合并排列后的阶段处理结果，得到合并结果。

22、通过根据执行依赖关系确定处理层级值，并按照处理层级值和阶段标识排列阶段处理结果，可以形成确定的结果合并次序，减少不同计算节点返回结果的先后顺序对合并次序造成的影响。

23、进一步地，在步骤s6中，更新物联网设备的云端状态后，读取物联网设备的预警状态和预警起始时间，预警状态包括未触发状态和触发状态；当预警状态为未触发状态、未记录预警起始时间且合并结果满足预警条件时，记录合并结果对应时间窗口的起始时间作为预警起始时间；当预警状态为未触发状态且后续相邻时间窗口的合并结果不满足预警条件时，清除预警起始时间；当预警状态为未触发状态、后续相邻时间窗口的合并结果连续满足预警条件，并且当前时间窗口的结束时间与预警起始时间之间的时长达到预设持续时长时，生成预警信息并将预警状态设置为触发状态；当预警状态为触发状态且后续相邻时间窗口的合并结果不满足预警条件时，生成恢复信息，将预警状态设置为未触发状态，并清除预警起始时间。

24、通过结合相邻时间窗口的合并结果和预设持续时长判断预警触发条件，可以降低短时数据变化直接触发预警的可能性。根据后续合并结果改变预警状态并生成恢复信息，可以使预警状态与设备数据的后续变化相对应。

25、本技术实施例的第二方面，提供了一种基于云边协同的物联网数据实时处理系统，该系统包括：数据接入网关，用于与物联网设备和云端流式数据通道通信，接收物联网设备原始数据，按照设备模型生成包含设备标识、采集时间和测点数据的标准化数据，将标准化数据写入云端流式数据通道，并根据控制指令调节数据写入速率；云端流式数据通道，用于缓存标准化数据，向任务生成模块提供标准化数据，并向流量控制模块提供缓存队列长度；任务生成模块，用于按采集时间将标准化数据划分为时间窗口，生成带任务标识、任务优先级和规则版本号的数据处理任务，将数据处理任务划分为具有执行依赖关系的多个处理阶段，并将数据处理任务和多个处理阶段提供给版本管理模块和调度模块；版本管理模块，用于读取云端当前规则版本号和首个处理阶段的执行状态，在云端当前规则版本号改变时，仅更新首个处理阶段尚未开始执行的数据处理任务的规则版本号，并保持首个处理阶段已经开始执行的数据处理任务的规则版本号以及云端计算节点加载的数据处理任务所使用的规则版本号，将保持或更新后的规则版本号提供给调度模块；流量控制模块，用于根据缓存队列长度和预设速率映射关系确定目标写入速率，生成控制指令并将控制指令发送至数据接入网关；调度模块，用于根据任务优先级和等待时长生成任务处理顺序，根据数据处理任务的规则版本号、云端计算节点的已加载规则版本号和节点负载值确定各处理阶段的目标计算节点，并按照任务处理顺序，将首个处理阶段和前序处理阶段已完成的后续处理阶段发送至对应的目标计算节点；多个云端计算节点，用于执行调度模块发送的处理阶段，向调度模块提供节点负载值、已加载规则版本号和处理阶段的执行状态，向版本管理模块提供首个处理阶段的执行状态，并将生成的阶段处理结果发送至结果处理模块；结果处理模块，用于在携带相同任务标识的全部处理阶段完成后，合并对应的阶段处理结果，并根据合并结果更新对应物联网设备的云端状态。

26、通过数据接入网关和云端流式数据通道传递及缓存标准化数据，可以将设备原始数据转换为供云端处理的数据形式。流量控制模块根据缓存队列长度生成控制指令，可以使数据写入速率随队列积压程度进行调整。任务生成模块、版本管理模块和调度模块传递任务信息、规则版本号和处理阶段，可以使处理阶段被发送至规则版本匹配的云端计算节点，并结合节点负载值确定具体的执行节点。云端计算节点反馈处理阶段的执行状态，使调度模块能够在前序处理阶段完成后发送对应的后续处理阶段。结果处理模块在全部处理阶段完成后合并处理结果并更新设备云端状态，使阶段执行结果能够衔接至设备状态更新过程。由此，有助于使数据接入、流量调节、任务调度、阶段执行和状态更新形成相互衔接的物联网实时数据处理过程。