用于物联网数据批流一体处理的系统及方法与流程

1.本技术涉及数据处理
技术领域：
：，具体地涉及一种用于物联网数据批流一体处理的系统及方法。
背景技术：
：：2.在工业互联网领域，随着万物互联和人机交互等工业场景的出现，海量工业设备需要通过互联网和5g等手段接入到云平台，进而将大量物联网数据实时地从边缘侧传输到云平台，或者将操作指令通过云平台发送到边缘端，这些传输到云平台的数据需要经过各种不同形式的统计分析和数据挖掘，进而赋能业务，指导运营决策。3.为了满足不同场景下的海量物联网数据分析需求，目前行业内的通用做法是通过引入不同的大数据组件来解决不同场景下的分析需求。例如：按天或按月的统计分析采用spark计算引擎+hive数仓的分析模式；记录存储工业设备最新工况数据采用canssandra非关系型数据库+elasticsearch索引数据库；存储工业设备的历史工况数据采用hbase非关系型数据库；实时分析设备的物联网数据则采用flink实时计算引擎+消息队列kafka+mpp数据库dorisdb等多种支持实时分析需求的数据存储系统。如此，整个数据分析过程会涉及到种类纷繁复杂的分布式数据存储系统和分布式计算引擎，而运行并维护这样一套大数据分析系统需要很大的技术投入，具有很高的运维成本和技术复杂度，中小企业一般无法维护和运营这样一套复杂的大数据系统。同时，同一份物联网数据为了满足不同场景下的数据分析需求需要在不同存储系统重复存储，造成了很大的存储浪费及数据的不一致性等问题，目前针对此类场景、暂没有很好的解决方案。4.因此，现有技术中所采用的海量物联网数据分析的架构设计复杂，运维成本和经济成本均较高，且存在可能导致数据不一致的风险的问题。技术实现要素：5.本技术实施例的目的是提供一种用于物联网数据批流一体处理的系统及方法，用以解决现有技术中所采用的海量物联网数据分析的架构设计复杂，运维成本和经济成本均较高，且存在可能导致数据不一致的风险的问题。6.为了实现上述目的，本技术第一方面提供一种用于物联网数据批流一体处理的系统，该系统包括：7.云平台，被配置成接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列；8.flink计算引擎，分别与云平台和数据湖通信，被配置成对kafka消息队列中的消息数据进行数据处理并存储至数据湖中，以及根据数据分析需求获取数据湖中设定时刻的快照信息进行数据分析；9.数据湖，被配置成对数据处理过程中各阶段的数据进行分层存储。10.在本技术实施例中，flink计算引擎还被配置成：11.获取输入的指令参数；12.在指令参数为批处理指令参数的情况下，打开批处理接口以执行批处理逻辑；13.在指令参数为实时流处理指令参数的情况下，打开实时流处理接口以执行实时流处理逻辑。14.在本技术实施例中，该系统还包括查询引擎，与数据湖通信，被配置成根据查询需求查询数据湖中的数据；15.数据湖还被配置成将查询引擎的查询结果发送至工业应用。16.在本技术实施例中，数据湖包括数据缓冲层，flink计算引擎还被配置成：17.在数据缓冲层中新建用于存储数据的数据缓冲表；18.将kafka消息队列中的消息数据写入数据缓冲表中。19.在本技术实施例中，flink计算引擎还被配置成：20.读取数据缓冲表中的缓冲数据；21.基于数据湖中各数据层对应的存储规则以及预设指标，对缓冲数据进行逐层处理并存储。22.在本技术实施例中，数据湖还包括数据原始层、数据标准层、数据整合层、数据应用层和维度层，flink计算引擎还被配置成：23.对缓冲数据进行规整和结构化清洗后，存储至数据原始层；24.将数据原始层中的数据进行标准化；25.将完成标准化的数据存储至数据标准层；26.根据预设指标将数据标准层的数据进行轻度汇总，并进行主题划分后存入数据整合层；27.根据预设指标对数据整合层的数据、数据标准层的数据以及维度层的数据进行综合分析，得到分析结果；28.将分析结果存入数据应用层。29.本技术第二方面提供一种用于物联网数据批流一体处理的方法，应用于用于物联网数据批流一体处理的系统，系统包括云平台、数据湖和flink计算引擎，flink计算引擎分别与云平台和数据湖通信，该方法包括：30.通过云平台接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列；31.通过flink计算引擎对kafka消息队列中的消息数据进行数据处理并存储至数据湖中；32.通过数据湖将数据处理过程中各阶段的数据进行分层存储；33.通过flink计算引擎根据分析需求获取数据湖中设定时刻的快照文件进行数据分析。34.在本技术实施例中，该方法还包括：35.获取输入的指令参数；36.在指令参数为批处理指令参数的情况下，打卡批处理接口以执行批处理逻辑；37.在指令参数为实时流处理指令参数的情况下，打开实时流处理接口以执行实时流处理逻辑。38.在本技术实施例中，数据湖包括数据缓冲层，通过计算引擎对kafka消息队列中的消息数据进行数据处理并存储至数据湖中包括：39.在数据缓冲层中新建用于存储数据的数据缓冲表；40.将kafka消息队列中的消息数据写入数据缓冲表中；41.读取数据缓冲表中的缓冲数据；42.基于数据湖中各数据层对应的存储规则以及预设指标，对缓冲数据进行逐层处理并存储。43.在本技术实施例中，数据湖还包括数据原始层、数据标准层、数据整合层、数据应用层和维度层，基于数据湖中各数据层对应的存储规则以及预设指标，对缓冲数据进行逐层处理并存储，包括：44.对缓冲数据进行规整和结构化清洗后，存储至数据原始层；45.将数据原始层中的数据进行标准化；46.将完成标准化的数据存储至数据标准层；47.根据预设指标将数据标准层的数据进行轻度汇总，并进行主题划分后存入数据整合层；48.根据预设指标对数据整合层的数据、数据标准层的数据以及维度层的数据进行综合分析，得到分析结果；49.将分析结果存入数据应用层。50.通过上述技术方案，提供一种用于物联网数据批流一体处理的系统，该系统包括云平台、flink计算引擎和数据湖，flink计算引擎分别与云平台和数据湖通信；云平台接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列；flink计算引擎对kafka消息队列中的消息数据进行数据处理并存储至数据湖中，以及根据数据分析需求获取数据湖中设定时刻的快照信息进行数据分析；数据湖对数据处理过程中各阶段的数据进行分层存储；其中，数据湖包括数据缓冲层、数据原始层、数据标准层、数据整合层、数据应用层、临时层和维度层。本技术通过整合flink计算引擎和数据湖，基于数据湖的快照机制以及flink对有界无界数据流的支持，实现了物联网数据的集中存储和统一计算，打破了不同部门不同系统之间的信息孤岛，有效保障了数据一致性，提高了数据质量，与传统物联网数据分析系统的架构设计相比，极大降低了系统的复杂度。51.本技术实施例的其它特征和优点将在随后的具体实施方式部分予以详细说明。附图说明52.附图是用来提供对本技术实施例的进一步理解，并且构成说明书的一部分，与下面的具体实施方式一起用于解释本技术实施例，但并不构成对本技术实施例的限制。在附图中：53.图1为本技术实施例提供的一种用于物联网数据批流一体处理的系统的结构示意图；54.图2为本技术一具体实施例提供的数据湖中数据分层处理的流程示意图；55.图3为本技术一具体实施例提供的flink计算引擎调度流程示意图；56.图4为本技术实施例提供的一种用于物联网数据批流一体处理的方法的流程示意图。57.附图标记说明58.10ꢀꢀꢀꢀꢀꢀ云平台ꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀ20ꢀꢀꢀꢀꢀꢀflink计算引擎59.30ꢀꢀꢀꢀꢀꢀ数据湖ꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀ40ꢀꢀꢀꢀꢀꢀ查询引擎60.100ꢀꢀꢀꢀꢀ数据缓冲层ꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀ200ꢀꢀꢀꢀꢀ数据原始层61.300ꢀꢀꢀꢀꢀ数据标准层ꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀ400ꢀꢀꢀꢀꢀ数据整合层62.500ꢀꢀꢀꢀꢀ数据应用层ꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀ600ꢀꢀꢀꢀꢀ维度层63.700ꢀꢀꢀꢀꢀ临时层ꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀꢀ800ꢀꢀꢀꢀꢀ任务调度监控系统64.900ꢀꢀꢀꢀꢀ元数据管理数据字典具体实施方式65.为使本技术实施例的目的、技术方案和优点更加清楚，下面将结合本技术实施例中的附图，对本技术实施例中的技术方案进行清楚、完整地描述，应当理解的是，此处所描述的具体实施方式仅用于说明和解释本技术实施例，并不用于限制本技术实施例。基于本技术中的实施例，本领域普通技术人员在没有做出创造性劳动的前提下所获得的所有其他实施例，都属于本技术保护的范围。66.需要说明，若本技术实施例中有涉及方向性指示(诸如上、下、左、右、前、后……)，则该方向性指示仅用于解释在某一特定姿态(如附图所示)下各部件之间的相对位置关系、运动情况等，如果该特定姿态发生改变时，则该方向性指示也相应地随之改变。67.另外，若本技术实施例中有涉及“第一”、“第二”等的描述，则该“第一”、“第二”等的描述仅用于描述目的，而不能理解为指示或暗示其相对重要性或者隐含指明所指示的技术特征的数量。由此，限定有“第一”、“第二”的特征可以明示或者隐含地包括至少一个该特征。另外，各个实施例之间的技术方案可以相互结合，但是必须是以本领域普通技术人员能够实现为基础，当技术方案的结合出现相互矛盾或无法实现时应当认为这种技术方案的结合不存在，也不在本技术要求的保护范围之内。68.图1为本技术实施例提供的一种用于物联网数据批流一体处理的系统的结构示意图。如图1所示，本技术实施例提供一种用于物联网数据批流一体处理的系统，该系统可以包括：69.云平台10，被配置成接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列；70.flink计算引擎20，分别与云平台10和数据湖30通信，被配置成对kafka消息队列中的消息数据进行数据处理并存储至数据湖30中，以及根据数据分析需求获取数据湖30中设定时刻的快照信息进行数据分析；71.数据湖30，被配置成对数据处理过程中各阶段的数据进行分层存储；72.其中，数据湖30包括数据缓冲层、数据原始层、数据标准层、数据整合层、数据应用层、临时层和维度层。73.为了简化目前海量物联网数据分析的复杂架构设计，降低运维成本和经济成本，提高数据的质量，避免对系统重复存储造成数据一致性问题。本技术实施例提供了一种用于物联网数据批流一体处理的系统，该系统可以包括云平台10、flink计算引擎20和数据湖30，云平台10与flink计算引擎20通信，数据湖30与flink计算引擎20通信。74.在本技术实施例中，云平台10一般指工业互联网平台，云平台10可以接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列中。其中，边缘侧物联网设备包括车辆、产线设备、仪器仪表和传感器等。边缘侧物联网设备上传的工作数据即物联网数据，物联网数据可以包括工程机械设备的工况数据、工程机械设备的地理位置数据、产线生产设备的工况数据、工业园区的水电气表数据、工业园区的门禁数据和摄像头数据等。在一个示例中，边缘侧物联网设备会实时通过5g网络上传工作数据到云平台10的网关侧，云平台10的网关侧则负责将工作数据转发至kafka消息队列中。进一步地，将kafka消息队列中的数据存入数据湖30中。75.在本技术实施例中，由于flink计算引擎20具备基于事件维度进行数据处理的能力，即flink计算引擎20一方面可以处理有界数据流，另一方面支持无界数据流的处理。因此，本技术实施例中选择了flink计算引擎20作为分布式计算引擎来对物联网数据进行处理。具体地，flink计算引擎20分别与云平台10和数据湖30通信，flink计算引擎20可以对kafka消息队列中的消息数据进行数据处理，并存储至数据湖30中。在一个示例中，flink计算引擎20一方面可以通过kafka-connector对接kafka消息队列，另一方面通过iceberg-connector连接数据湖30，并通过flinksql将海量物联网数据写入数据湖30。76.在本技术实施例中，数据湖30(iceberg)作为一种用于大型数据分析场景的开放表格式(tableformat)，它使用一种类似于sql表的高性能表格式，iceberg格式表单表可以存储数十pb数据，可以适配不同计算引擎，提供高性能的读写和元数据管理功能，支持实时/批量数据写入和读取，支持事务acid，支持添加、删除和更新数据，不绑定任何底层存储，支持parquet、orc和avro格式兼容行存储和列存储，支持快照数据重复查询，具备版本回滚功能，同时，数据湖30还是一种无服务的工作模式。因此，本技术实施例创新性地将flink计算引擎20和数据湖30结合，利用数据湖30对数据处理过程中各阶段的数据进行分层存储。77.在本技术实施例中，为了使所有数据能够有秩序地流转，数据的整个生命周期能够清晰明确地被感知，本技术实施例根据物联网数据的特点设计了数据湖30的数据分层，以使得数据结构更加清晰且便于后期维护。具体地，数据湖30的数据分层可以包括数据缓冲层、数据原始层、数据标准层、数据整合层、数据应用层、临时层和维度层。78.其中，数据缓冲层用于暂时存储kafka消息队列中的原始非结构化的物联网数据。数据原始层用于存储经数据规整和结构化清洗之后的原始数据。数据标准层用于存储对数据原始层的原始数据进行标准化的数据清洗后得到的有价值的业务数据。数据整合层用于存储对数据标准层的业务数据进行轻度汇总和主题计算后的数据。数据应用层用于存储对数据整合层的数据根据不同主题进行深度管理分析和指标分类汇总后得到的不同主题类型的统计数据，一般为宽表数据。此外，临时层对数据进行处理并逐层存储的过程中所产生的中间数据，以便后续调用。维度层用于存储上述不同数据层进行数据分析时均需用到的公共数据。79.如此，通过数据湖的分层设计，可以清楚地界定物联网数据与维度数据的逻辑关系，减少了重复的开发工作，数据处理流程清晰，结构明确，功能明确。80.在本技术实施例中，flink计算引擎20还可以根据数据分析需求获取数据湖30中设定时刻的快照信息进行数据分析。其中，数据分析需求包括批量处理的数据分析和实时处理的数据分析。具体地，基于数据湖30时间旅行的特性，数据湖30在不同时刻保存了数据文件的快照(snapshot)信息，这些快照信息记录了某个时刻的所有数据文件列表，并且随着时间的推移，前面时刻的snapshot会包含上一时刻的snapshot所记录的所有数据文件信息。因此，可以根据snapshot的时间特性去读取任意时间段内数据。同时，由于当前时刻的数据文件快照snapshot-1包含前一时刻的数据文件快照snapshot-0的全部信息，那么snapshot-1与snapshot-0之间的增量部分即为最新数据，通过调控iceberg快照记录的时间周期，我们可以获取到最近(0-30min)内的任意最新数据，对最近的增量数据进行分析，便可得到近实时的分析结果。81.在本技术实施例中，可以根据数据湖30的这种特性一方面进行批量处理的数据分析，即按天或按月的数据统计分析，每次只需要根据数据湖30的snapshot读取一天或者一个月时间跨度的数据进行分析即可；另一方面，还可以同时获取到最新的实时分析数据结果。在本技术实施例中，为了提高计算效率，提高数据分析的时效性，节约计算资源和存储资源，我们选择了flink分布式计算引擎进行数据计算，因为flink在设计之初既同时具备对有界数据和无界数据进行分析处理的能力，即flink集成了处理有界和无界数据流的各种算子、函数和方法。因此，本技术实施例中，我们将数据计算引擎flink和数据存储系统iceberg数据湖30在批量和实时数据处理方面的特性进行有效的整合和创新性的流程设计，按照iceberg数据湖30中数据分层进行物联网数据的离线分析和实时分析，极大降低了海量物联网数据处理系统的复杂度和成本，同时提高了数据的分析效率。82.通过上述技术方案，提供一种用于物联网数据批流一体处理的系统，该系统包括云平台、flink计算引擎和数据湖，flink计算引擎分别与云平台和数据湖通信；云平台接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列；flink计算引擎对kafka消息队列中的消息数据进行数据处理并存储至数据湖中，以及根据数据分析需求获取数据湖中设定时刻的快照信息进行数据分析；数据湖对数据处理过程中各阶段的数据进行分层存储；其中，数据湖包括数据缓冲层、数据原始层、数据标准层、数据整合层、数据应用层、临时层和维度层。本技术通过整合flink计算引擎和数据湖，基于数据湖的快照机制以及flink对有界无界数据流的支持，实现了物联网数据的集中存储和统一计算，打破了不同部门不同系统之间的信息孤岛，有效保障了数据一致性，提高了数据质量，与传统物联网数据分析系统的架构设计相比，极大降低了系统的复杂度。83.在本技术实施例中，flink计算引擎20还可以被配置成：84.获取输入的指令参数；85.在指令参数为批处理指令参数的情况下，打开批处理接口以执行批处理逻辑；86.在指令参数为实时流处理指令参数的情况下，打开实时流处理接口以执行实时流处理逻辑。87.在本技术实施例中，对有界数据流的处理通常被称为批处理，批处理不需要有序地获取数据，在批处理模式下，首先按将数据流持久化到存储系统中，然后对整个数据集的数据进行读取、排序、统计或汇总计算，最后输出结果。对无界数据流的通常在数据生成时进行实时处理，称为实时流处理，因为无界数据流的数据输入是无限的，所以必须持续地处理。在本技术实施例中，flink在设计之初既同时具备对有界数据和无界数据进行分析处理的能力。具体地，先获取输入的指令参数，指令参数可以保留批处理质量参数和实时流处理指令参数，该指令参数可以由技术开发人员根据数据分析需求通过编写代码传入。进一步地，在指令参数为批处理指令参数的情况下，可以打开flink计算引擎20的批处理接口以执行批处理逻辑；在指令参数为实时流处理指令参数的情况下，可以打开flink计算引擎20的实时流处理接口以执行实时流处理逻辑。如此，通过本系统既可以首先按实时分析又可以实现离线分析，离线分析和实时分析共享相同的存储系统和计算资源，降低了运维成本；并且，离线分析和实时分析公用一套分析代码，降低了开发成本，提高了开发效率。88.在本技术实施例中，该系统还可以包括查询引擎40，与数据湖30通信，被配置成根据查询需求查询数据湖30中的数据；89.数据湖30还可以被配置成将查询引擎40的查询结果发送至工业应用。90.在本技术实施例中，该系统还可以包括查询引擎40，查询引擎40与数据湖30通信，可以根据查询需求查询数据湖30中的数据。进一步地，数据湖30可以将查询引擎40的查询结果返回给对应的工业应用。其中，查询引擎40可以为数据查询引擎40presto。具体地，物联网数据经过逐层分析处理之后，最终的各种统计指标会以大宽表的形式存储在数据湖30iceberg的数据应用层500，为了进一步简化物联网数据统计分析的大数据架构，本技术实施例并没有选择将数据导出到外部存储系统，而是通过prestosql直接查询iceberg中的统计分析结果，通过在prestosql安装部署目录etc/catalog下增加iceberg的相关配置，实现了prestosql与iceberg的打通，即通过查询引擎40可以直接查询数据湖30iceberg中的数据，用户通过prestosql提供的jdbc连接地址对数据湖30iceberg中的数据进行快速的查询和计算，将结果直接从数据湖30返回给工业应用。91.图2为本技术一具体实施例提供的数据湖中数据分层处理的流程示意图。如图2所示，物联网数据写入数据湖30进入数据分析，经过数据缓冲层100、数据原始层200、数据标准层300、数据整合层400以及数据应用层500后实现数据应用，其中数据湖中还包括有维度层600和临时层700，通过任务调度监控系统800和元数据管理数据字典900实现各层之间的数据处理任务调度与数据处理。通过对数据湖30的分层设计，清楚地界定了业务和财务类的数据的逻辑关系和应用范围，减少了很多重复的开发工作，数据处理流程清晰，结构明确，功能明确。92.在本技术实施例中，flink计算引擎20还被可以配置成：93.在数据缓冲层100中新建用于存储数据的数据缓冲表；94.将kafka消息队列中的消息数据写入数据缓冲表中。95.在本技术实施例中，数据缓冲表是指数据湖30的缓冲数据层新建的mor表(mergeonread)。根据数据特点，可对数据湖30进行数据分层，其中数据缓冲层100(bdl，bufferdatalayer)主要用来暂时存储flink写入的原始数据，bdl层数据的有效期比较短，如1-7天，有效期后之后需要删除。96.在本技术实施例中，flink计算引擎20还可以被配置成：97.读取数据缓冲表中的缓冲数据；98.基于数据湖30中各数据层对应的存储规则以及预设指标，对缓冲数据进行逐层处理并存储。99.在本技术实施例中，数据分层为根据数据特点对数据湖30设计的分层，数据分层对应的存储规则包括对存入该数据分层的数据的存储形式。数据分层互相之间存在一定的依赖关系，对数据进行处理将按照业务逻辑和各数据分层间的依赖关系进行逐层处理。100.在本技术实施例中，flink计算引擎20还可以被配置成：101.对缓冲数据进行规整和结构化清洗后，存储至数据原始层200；102.将数据原始层200中的数据进行标准化；103.将完成标准化的数据存储至数据标准层300；104.根据预设指标将数据标准层300的数据进行轻度汇总，并进行主题划分后存入数据整合层400；105.根据预设指标对数据整合层400的数据、数据标准层300的数据以及维度层600的数据进行综合分析，得到分析结果；106.将分析结果存入数据应用层500。107.具体地，flink计算引擎20将非结构化的缓冲数据进行规整和结构化处理之后存储到数据原始层200。在一个示例中，可以先创建数据原始层200各业务对应的iceberg的表，然后通过flink计算引擎20批量读取数据缓冲层100的iceberg表中的各业务数据，然后通过flink计算引擎20批量写入数据原始层200的iceberg表中，本步骤的主要开发语言是sql，通过编写数据结构化处理的代码，然后运行代码，达到对数据的简单清晰和结构化处理的目的，通过这一步操作，我们就得到了保持原始数据所有特征的所有结构化的业务数据。108.进一步地，由于iceberg支持根据快照进行全量或者增量读取数据，所以第一次分析可以采用快照读取全部数据经过分析然后存储到数据原始层200，之后通过一些iceberg参数的配置，就可以通过增量读取的方式进行增量分析。数据原始层200的数据准备好之后，可以继续创建数据标准层300的iceberg分区表，然后通过flink计算引擎20读取数据原始层200的数据，通过sql编码，实现清洗，去除脏数据、不合要求的数据和过期数据，然后将代码在flink计算引擎20上进行运行，实现将数据原始层200的数据批量写入到数据标准层300，按照iceberg数据湖30分层的原则，数据标准层300需要保留历史数据和当前的最新数据，即全量数据。109.在一个示例中，由于物联网数据时刻都在更新，为了充分发挥数据的价值，我们需要同时保存物联网设备的历史数据和最新工况数据。对此，传统方法是每天做一个数据分片，这个分片记录当天之前的全量历史数据，每天都会有一个分片，时间长了之后就会在数据库中存在大量的历史重复数据，占用存储资源和计算资源。本技术实施例中，充分利用了iceberg表数据可以进行行级别更新和实时插入(即读写分离)的功能，我们在数据标准差层只需要一个数据分片来存储当前和历史的全量数据，当有数据变化时，直接通过更新操作即可完成数据的更新和插入，无需重新创建数据分区和分片，同时我们也可以得到增量分析的结果，即获取了实时分析的结果。如此，可以提高对系统存储资源和计算资源的充分利用。110.进一步地，在数据标准层300的数据更新完成之后，需要将数据标准层300的数据做轻度汇总后批量写入数据整合层400。具体地，先创建数据整合层400的建表语句，完成数据整合层400iceberg表的创建之后通过flink计算引擎20读取数据标准层300的业务表数据，再通过flink计算引擎20批量写入数据整合层400，以完成各业务数据的轻度汇总和不同主题的划分。111.进一步地，flink计算引擎20可以根据预设指标对数据整合层400的数据、数据标准层300的数据以及维度层600的数据进行综合分析，得到分析结果，并将分析结果存入数据应用层500。具体地，先创建数据应用层500的iceberg表，然后通过flink计算引擎20根据不同的业务逻辑，对维度层600的维度数据、数据整合层400的轻度汇总数据和数据标准层300的明细数据，进行综合的处理和分析，然后将结果批量写入数据应用层500对应的iceberg表中。112.在本技术实施例中，由于宽表的加工过程会设及到很多张表，会加载很多历史数据进行综合处理分析，一般会占用大量的计算资源。因此本技术实施例中我们只在第一次会进行全量数据的分析，做一次初始化操作之后，后续所有的分析只会进行增量数据的分析，充分利用iceberg可以进行增量读数据的优势，将增量分析的结果以追加写的模式合并更新到之前的历史分析结果中，确保分析结果的全面性，同时我们也可以对增量分析的结果(实时分析)进行备份保存和更新，这样既能得到批量分析的结果也可得到实时分析的结果；只做增量数据的分析之后就会节省大量的计算资源和存储资源。按照这样的操作之后就可以得到物联网数据的实时和批量处理分析统计结果。113.图3为本技术一具体实施例提供的flink计算引擎调度流程示意图。如图3所示，为了达到自动集成多系统前端业务数据，实时体现业务实际，及时发现业务异常的目标，flink计算引擎20在数据湖30的不同层之间(bdl—》odl—》sdl—》idl—》adl)的数据处理任务按照相互依赖关系，按照一定的频率定时的调度起来。按照一定的频率定时的调度起来，可以让机器自动去进行物联网数据的运算，大大提高了数据开发效率。114.图4为本技术实施例提供的一种用于物联网数据批流一体处理的方法的流程示意图。如图4所示，本技术实施例提供一种用于物联网数据批流一体处理的方法，应用于用于物联网数据批流一体处理的系统，系统包括云平台、数据湖和flink计算引擎，flink计算引擎分别与云平台和数据湖通信，该方法可以包括下列步骤：115.步骤101、通过云平台接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列；116.步骤102、通过flink计算引擎对kafka消息队列中的消息数据进行数据处理并存储至数据湖中；117.步骤103、通过数据湖将数据处理过程中各阶段的数据进行分层存储；118.步骤104、通过flink计算引擎根据分析需求获取数据湖中设定时刻的快照文件进行数据分析；119.其中，数据湖包括数据缓冲层、数据原始层、数据标准层、数据整合层、数据应用层、临时层和维度层。120.在本技术实施例中，云平台一般指工业互联网平台，云平台可以接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列中。其中，边缘侧物联网设备包括车辆、产线设备、仪器仪表和传感器等。在一个示例中，边缘侧物联网设备会实时通过5g网络上传工作数据到云平台的网关侧，云平台的网关侧则负责将工作数据转发至kafka消息队列中。进一步地，将kafka消息队列中的数据存入数据湖中。121.在本技术实施例中，flink计算引擎分别与云平台和数据湖通信，flink计算引擎可以对kafka消息队列中的消息数据进行数据处理，并存储至数据湖中。在一个示例中，flink计算引擎一方面可以通过kafka-connector对接kafka消息队列，另一方面通过iceberg-connector连接数据湖，并通过flinksql将海量物联网数据写入数据湖。进一步地，本技术实施例创新性地将flink计算引擎和数据湖结合，利用数据湖对数据处理过程中各阶段的数据进行分层存储。122.在本技术实施例中，为了使所有数据能够有秩序地流转，数据的整个生命周期能够清晰明确地被感知，本技术实施例根据物联网数据的特点设计了数据湖iceberg的数据分层，以使得数据结构更加清晰且便于后期维护。具体地，数据湖的数据分层可以包括数据缓冲层、数据原始层、数据标准层、数据整合层、数据应用层、临时层和维度层。如此，通过数据湖的分层设计，可以清楚地界定物联网数据与维度数据的逻辑关系，减少了重复的开发工作，数据处理流程清晰，结构明确，功能明确。123.在本技术实施例中，flink计算引擎还可以根据数据分析需求获取数据湖中设定时刻的快照信息进行数据分析。具体地，基于数据湖iceberg时间旅行的特性，数据湖在不同时刻保存了数据文件的快照(snapshot)信息，这些快照信息记录了某个时刻的所有数据文件列表，并且随着时间的推移，前面时刻的snapshot会包含上一时刻的snapshot所记录的所有数据文件信息。因此，可以根据snapshot的时间特性去读取任意时间段内数据。同时，由于当前时刻的数据文件快照snapshot-1包含前一时刻的数据文件快照snapshot-0的全部信息，那么snapshot-1与snapshot-0之间的增量部分即为最新数据，通过调控iceberg快照记录的时间周期，我们可以获取到最近(0-30min)内的任意最新数据，对最近的增量数据进行分析，便可得到近实时的分析结果。124.在本技术实施例中，可以根据数据湖的这种特性一方面进行批量处理的数据分析，即按天或按月的数据统计分析，每次只需要根据数据湖的snapshot读取一天或者一个月时间跨度的数据进行分析即可；另一方面，还可以同时获取到最新的实时分析数据结果。在本技术实施例中，可以将数据计算引擎flink和数据存储系统iceberg数据湖在批量和实时数据处理方面的特性进行有效的整合和创新性的流程设计，按照iceberg数据湖中数据分层进行物联网数据的离线分析和实时分析，极大降低了海量物联网数据处理系统的复杂度和成本，同时提高了数据的分析效率。125.在本技术实施例中，该方法还可以包括：126.获取输入的指令参数；127.在指令参数为批处理指令参数的情况下，打卡批处理接口以执行批处理逻辑；128.在指令参数为实时流处理指令参数的情况下，打开实时流处理接口以执行实时流处理逻辑。129.在本技术实施例中，flink在设计之初既同时具备对有界数据和无界数据进行分析处理的能力。具体地，先获取输入的指令参数，指令参数可以保留批处理质量参数和实时流处理指令参数，该指令参数可以由技术开发人员根据数据分析需求通过编写代码传入。进一步地，在指令参数为批处理指令参数的情况下，可以打开flink计算引擎的批处理接口以执行批处理逻辑；在指令参数为实时流处理指令参数的情况下，可以打开flink计算引擎的实时流处理接口以执行实时流处理逻辑。如此，通过本系统既可以首先按实时分析又可以实现离线分析，离线分析和实时分析共享相同的存储系统和计算资源，降低了运维成本；并且，离线分析和实时分析公用一套分析代码，降低了开发成本，提高了开发效率。130.在本技术实施例中，步骤102、通过计算引擎对kafka消息队列中的消息数据进行数据处理并存储至数据湖中可以包括：131.在数据缓冲层中新建用于存储数据的数据缓冲表；132.将kafka消息队列中的消息数据写入数据缓冲表中；133.读取数据缓冲表中的缓冲数据；134.基于数据湖中各数据层对应的存储规则以及预设指标，对缓冲数据进行逐层处理并存储。135.在本技术实施例中，数据缓冲表是指数据湖的缓冲数据层新建的mor表(mergeonread)。根据数据特点，可对数据湖进行数据分层，其中数据缓冲层(bdl，bufferdatalayer)主要用来暂时存储flink写入的原始数据，bdl层数据的有效期比较短，如1-7天，有效期后之后需要删除。在本技术实施例中，数据分层为根据数据特点对数据湖设计的分层，数据分层对应的存储规则包括对存入该数据分层的数据的存储形式。数据分层互相之间存在一定的依赖关系，对数据进行处理将按照业务逻辑和各数据分层间的依赖关系进行逐层处理。136.在本技术实施例中，基于数据湖中各数据层对应的存储规则以及预设指标，对缓冲数据进行逐层处理并存储，可以包括：137.对缓冲数据进行规整和结构化清洗后，存储至数据原始层；138.将数据原始层中的数据进行标准化；139.将完成标准化的数据存储至数据标准层；140.根据预设指标将数据标准层的数据进行轻度汇总，并进行主题划分后存入数据整合层；141.根据预设指标对数据整合层的数据、数据标准层的数据以及维度层的数据进行综合分析，得到分析结果；142.将分析结果存入数据应用层。143.具体地，flink计算引擎将非结构化的缓冲数据进行规整和结构化处理之后存储到数据原始层。在一个示例中，可以先创建数据原始层各业务对应的iceberg的表，然后通过flink计算引擎批量读取数据缓冲层的iceberg表中的各业务数据，然后通过flink计算引擎批量写入数据原始层的iceberg表中，本步骤的主要开发语言是sql，通过编写数据结构化处理的代码，然后运行代码，达到对数据的简单清晰和结构化处理的目的，通过这一步操作，我们就得到了保持原始数据所有特征的所有结构化的业务数据。144.进一步地，由于iceberg支持根据快照进行全量或者增量读取数据，所以第一次分析可以采用快照读取全部数据经过分析然后存储到数据原始层，之后通过一些iceberg参数的配置，就可以通过增量读取的方式进行增量分析。数据原始层的数据准备好之后，可以继续创建数据标准层的iceberg分区表，然后通过flink计算引擎读取数据原始层的数据，通过sql编码，实现清洗，去除脏数据、不合要求的数据和过期数据，然后将代码在flink计算引擎上进行运行，实现将数据原始层的数据批量写入到数据标准层，按照iceberg数据湖分层的原则，数据标准层需要保留历史数据和当前的最新数据，即全量数据。145.在一个示例中，由于物联网数据时刻都在更新，为了充分发挥数据的价值，我们需要同时保存物联网设备的历史数据和最新工况数据。对此，传统方法是每天做一个数据分片，这个分片记录当天之前的全量历史数据，每天都会有一个分片，时间长了之后就会在数据库中存在大量的历史重复数据，占用存储资源和计算资源。本技术实施例中，充分利用了iceberg表数据可以进行行级别更新和实时插入(即读写分离)的功能，我们在数据标准差层只需要一个数据分片来存储当前和历史的全量数据，当有数据变化时，直接通过更新操作即可完成数据的更新和插入，无需重新创建数据分区和分片，同时我们也可以得到增量分析的结果，即获取了实时分析的结果。如此，可以提高对系统存储资源和计算资源的充分利用。146.进一步地，在数据标准层的数据更新完成之后，需要将数据标准层的数据做轻度汇总后批量写入数据整合层。具体地，先创建数据整合层的建表语句，完成数据整合层iceberg表的创建之后通过flink计算引擎读取数据标准层的业务表数据，再通过flink计算引擎批量写入数据整合层，以完成各业务数据的轻度汇总和不同主题的划分。147.进一步地，flink计算引擎可以根据预设指标对数据整合层的数据、数据标准层的数据以及维度层的数据进行综合分析，得到分析结果，并将分析结果存入数据应用层。具体地，先创建数据应用层的iceberg表，然后通过flink计算引擎根据不同的业务逻辑，对维度层的维度数据、数据整合层的轻度汇总数据和数据标准层的明细数据，进行综合的处理和分析，然后将结果批量写入数据应用层对应的iceberg表中。148.在本技术实施例中，由于宽表的加工过程会设及到很多张表，会加载很多历史数据进行综合处理分析，一般会占用大量的计算资源。因此本技术实施例中我们只在第一次会进行全量数据的分析，做一次初始化操作之后，后续所有的分析只会进行增量数据的分析，充分利用iceberg可以进行增量读数据的优势，将增量分析的结果以追加写的模式合并更新到之前的历史分析结果中，确保分析结果的全面性，同时我们也可以对增量分析的结果(实时分析)进行备份保存和更新，这样既能得到批量分析的结果也可得到实时分析的结果；只做增量数据的分析之后就会节省大量的计算资源和存储资源。按照这样的操作之后就可以得到物联网数据的实时和批量处理分析统计结果。149.通过上述技术方案，提供一种用于物联网数据批流一体处理的系统，该系统包括云平台、flink计算引擎和数据湖，flink计算引擎分别与云平台和数据湖通信；云平台接收边缘侧物联网设备上传的工作数据，并将工作数据存储至kafka消息队列；flink计算引擎对kafka消息队列中的消息数据进行数据处理并存储至数据湖中，以及根据数据分析需求获取数据湖中设定时刻的快照信息进行数据分析；数据湖对数据处理过程中各阶段的数据进行分层存储；其中，数据湖包括数据缓冲层、数据原始层、数据标准层、数据整合层、数据应用层、临时层和维度层。本技术通过整合flink计算引擎和数据湖，基于数据湖的快照机制以及flink对有界无界数据流的支持，实现了物联网数据的集中存储和统一计算，打破了不同部门不同系统之间的信息孤岛，有效保障了数据一致性，提高了数据质量，与传统物联网数据分析系统的架构设计相比，极大降低了系统的复杂度。150.本领域内的技术人员应明白，本技术的实施例可提供为方法、系统或计算机程序产品。因此，本技术可采用完全硬件实施例、完全软件实施例或结合软件和硬件方面的实施例的形式。而且，本技术可采用在一个或多个其中包含有计算机可用程序代码的计算机可用存储介质(包括但不限于磁盘存储器、cd-rom、光学存储器等)上实施的计算机程序产品的形式。151.本技术是参照根据本技术实施例的方法、设备(系统)和计算机程序产品的流程图和/或方框图来描述的。应理解可由计算机程序指令实现流程图和/或方框图中的每一流程和/或方框、以及流程图和/或方框图中的流程和/或方框的结合。可提供这些计算机程序指令到通用计算机、专用计算机、嵌入式处理机或其他可编程数据处理设备的处理器以产生一个机器，使得通过计算机或其他可编程数据处理设备的处理器执行的指令产生用于实现在流程图一个流程或多个流程和/或方框图一个方框或多个方框中指定的功能的装置。152.这些计算机程序指令也可存储在能引导计算机或其他可编程数据处理设备以特定方式工作的计算机可读存储器中，使得存储在该计算机可读存储器中的指令产生包括指令装置的制造品，该指令装置实现在流程图一个流程或多个流程和/或方框图一个方框或多个方框中指定的功能。153.这些计算机程序指令也可装载到计算机或其他可编程数据处理设备上，使得在计算机或其他可编程设备上执行一系列操作步骤以产生计算机实现的处理，从而在计算机或其他可编程设备上执行的指令提供用于实现在流程图一个流程或多个流程和/或方框图一个方框或多个方框中指定的功能的步骤。154.在一个典型的配置中，计算设备包括一个或多个处理器(cpu)、输入/输出接口、网络接口和内存。155.存储器可能包括计算机可读介质中的非永久性存储器，随机存取存储器(ram)和/或非易失性内存等形式，如只读存储器(rom)或闪存(flashram)。存储器是计算机可读介质的示例。156.计算机可读介质包括永久性和非永久性、可移动和非可移动媒体可以由任何方法或技术来实现信息存储。信息可以是计算机可读指令、数据结构、程序的模块或其他数据。计算机的存储介质的例子包括，但不限于相变内存(pram)、静态随机存取存储器(sram)、动态随机存取存储器(dram)、其他类型的随机存取存储器(ram)、只读存储器(rom)、电可擦除可编程只读存储器(eeprom)、快闪记忆体或其他内存技术、只读光盘只读存储器(cd-rom)、数字多功能光盘(dvd)或其他光学存储、磁盒式磁带，磁带磁盘存储或其他磁性存储设备或任何其他非传输介质，可用于存储可以被计算设备访问的信息。按照本文中的界定，计算机可读介质不包括暂存电脑可读媒体(transitorymedia)，如调制的数据信号和载波。157.还需要说明的是，术语“包括”、“包含”或者其任何其他变体意在涵盖非排他性的包含，从而使得包括一系列要素的过程、方法、商品或者设备不仅包括那些要素，而且还包括没有明确列出的其他要素，或者是还包括为这种过程、方法、商品或者设备所固有的要素。在没有更多限制的情况下，由语句“包括一个……”限定的要素，并不排除在包括要素的过程、方法、商品或者设备中还存在另外的相同要素。158.以上仅为本技术的实施例而已，并不用于限制本技术。对于本领域技术人员来说，本技术可以有各种更改和变化。凡在本技术的精神和原理之内所作的任何修改、等同替换、改进等，均应包含在本技术的权利要求范围之内。当前第1页12当前第1页12
技术特征：
1.一种用于物联网数据批流一体处理的系统，其特征在于，包括：云平台，被配置成接收边缘侧物联网设备上传的工作数据，并将所述工作数据存储至kafka消息队列；flink计算引擎，分别与所述云平台和数据湖通信，被配置成对所述kafka消息队列中的消息数据进行数据处理并存储至所述数据湖中，以及根据数据分析需求获取所述数据湖中设定时刻的快照信息进行数据分析；所述数据湖，被配置成对数据处理过程中各阶段的数据进行分层存储。2.根据权利要求1所述的系统，其特征在于，所述flink计算引擎还被配置成：获取输入的指令参数；在所述指令参数为批处理指令参数的情况下，打开批处理接口以执行批处理逻辑；在所述指令参数为实时流处理指令参数的情况下，打开实时流处理接口以执行实时流处理逻辑。3.根据权利要求1所述的系统，其特征在于，所述系统还包括查询引擎，与所述数据湖通信，被配置成根据查询需求查询所述数据湖中的数据；所述数据湖还被配置成将所述查询引擎的查询结果发送至工业应用。4.根据权利要求1所述的系统，其特征在于，所述flink计算引擎还被配置成：在所述数据缓冲层中新建用于存储数据的数据缓冲表；将所述kafka消息队列中的消息数据写入所述数据缓冲表中。5.根据权利要求4所述的系统，其特征在于，所述数据湖包括数据缓冲层，所述flink计算引擎还被配置成：读取所述数据缓冲表中的缓冲数据；基于所述数据湖中各数据层对应的存储规则以及预设指标，对所述缓冲数据进行逐层处理并存储。6.根据权利要求5所述的系统，其特征在于，所述数据湖还包括数据原始层、数据标准层、数据整合层、数据应用层和维度层，所述flink计算引擎还被配置成：对所述缓冲数据进行规整和结构化清洗后，存储至所述数据原始层；将所述数据原始层中的数据进行标准化；将完成标准化的数据存储至所述数据标准层；根据预设指标将所述数据标准层的数据进行轻度汇总，并进行主题划分后存入所述数据整合层；根据预设指标对所述数据整合层的数据、所述数据标准层的数据以及所述维度层的数据进行综合分析，得到分析结果；将所述分析结果存入所述数据应用层。7.一种用于物联网数据批流一体处理的方法，其特征在于，应用于用于物联网数据批流一体处理的系统，所述系统包括云平台、数据湖和flink计算引擎，所述flink计算引擎分别与所述云平台和所述数据湖通信，所述方法包括：通过所述云平台接收边缘侧物联网设备上传的工作数据，并将所述工作数据存储至kafka消息队列；通过所述flink计算引擎对所述kafka消息队列中的消息数据进行数据处理并存储至
所述数据湖中；通过所述数据湖将数据处理过程中各阶段的数据进行分层存储；通过所述flink计算引擎根据分析需求获取所述数据湖中设定时刻的快照文件进行数据分析。8.根据权利要求7所述的方法，其特征在于，所述方法还包括：获取输入的指令参数；在所述指令参数为批处理指令参数的情况下，打卡批处理接口以执行批处理逻辑；在所述指令参数为实时流处理指令参数的情况下，打开实时流处理接口以执行实时流处理逻辑。9.根据权利要求7所述的方法，其特征在于，所述数据湖包括数据缓冲层，所述通过计算引擎对所述kafka消息队列中的消息数据进行数据处理并存储至所述数据湖中包括：在所述数据缓冲层中新建用于存储数据的数据缓冲表；将所述kafka消息队列中的消息数据写入所述数据缓冲表中；读取所述数据缓冲表中的缓冲数据；基于所述数据湖中各数据层对应的存储规则以及预设指标，对所述缓冲数据进行逐层处理并存储。10.根据权利要求9所述的方法，其特征在于，所述数据湖还包括数据原始层、数据标准层、数据整合层、数据应用层和维度层，所述基于所述数据湖中各数据层对应的存储规则以及预设指标，对所述缓冲数据进行逐层处理并存储，包括：对所述缓冲数据进行规整和结构化清洗后，存储至所述数据原始层；将所述数据原始层中的数据进行标准化；将完成标准化的数据存储至所述数据标准层；根据预设指标将所述数据标准层的数据进行轻度汇总，并进行主题划分后存入所述数据整合层；根据预设指标对所述数据整合层的数据、所述数据标准层的数据以及所述维度层的数据进行综合分析，得到分析结果；将所述分析结果存入所述数据应用层。

技术总结
本申请公开了一种用于物联网数据批流一体处理的系统及方法。该系统包括：云平台被配置成接收边缘侧物联网设备上传的工作数据并存储至kafka消息队列；Flink计算引擎与云平台和数据湖通信，被配置成对kafka消息队列中的消息数据进行数据处理并存储至数据湖，以及根据数据分析需求获取数据湖中设定时刻的快照信息进行数据分析；数据湖被配置成对数据处理过程中各阶段的数据进行分层存储。本申请通过整合Flink计算引擎和数据湖，基于数据湖的快照机制及Flink对有界无界数据流的支持，实现了物联网数据的集中存储和统一计算，有效保障了数据一致性，降低了系统架构的复杂度。降低了系统架构的复杂度。降低了系统架构的复杂度。


技术研发人员：路培杰 曾光 刘文虎 周志忠
受保护的技术使用者：中科云谷科技有限公司
技术研发日：2023.06.29
技术公布日：2023/10/8