一种基于Kafka的工业企业实时泛在互联方法

摘要

本发明涉及一种基于Kafka的工业企业实时泛在互联方法，边缘计算服务通过Kafka Connect为Kafka提供工业物联网数据，企业管理信息系统通过KafkaConnect为Kafka提供管理信息系统数据，基于Kafka实现多层级数据的逐层上报，基于SchemaRegistry实现模式信息共享，最终实现工业企业的实时泛在互联。本发明的优点是：边缘计算服务通过KafkaConnect为Kafka提供了工业物联网数据；企业管理信息系统通过KafkaConnect为Kafka提供了管理信息系统数据；基于Kafka实现了多层级数据的逐层上报；基于Schema Registry实现了模式信息共享。

说明书

技术领域

本发明涉及工业物联网领域，尤其涉及一种基于Kafka的工业企业实时泛在互联方法。

背景技术

随着信息化的深入，工业企业的数据集成问题日益凸显。工业企业最大的两类数据来源是工业物联网数据和企业管理信息系统数据，而工业物联网数据来源于设备，企业管理信息系统数据来源于信息系统，两者之间如何实时、可靠的实现实时泛在连接成为解决问题的关键。

传统的基于ETL的企业数据集成，存在三方面问题：

1.实时性不足。传统ETL一般以天为单位进行数据同步,以牺牲数据实时性为代价，降低对目标数据库性能的影响。而要增加实时性，则必然增加读取频率，从而加重目标数据库的性能负担。

2.对工业物联网数据支持不足。钢铁企业设备量大，测点总量多，生产过程数据产生频率高，数据量巨大，时序数据是钢铁行业最主要的过程数据。时序数据的存储并不使用传统关系型数据库，传统的ETL方式无法支持工业物联网数据的集成。

3.对多层级数据逐层汇总支持不足。工业企业通常存在车间-产线-工厂-集团等多层级组织，各层级均有自己的信息系统。传统方式需要大量的开发工作去支持多层级的数据汇总，无法自动化的、低成本的实现数据的逐层汇总。

Kafka既是消息引擎系统，也是一个分布式流处理平台。支持多分区、多副本，具有很高的读写性能，很强的容错性和水平扩展能力，是目前广泛使用的企业数据总线。

Schema Registry通过注册表的方式在消息的生产者和消费者之间共享消息体的模式信息，这样就可以利用Avro、Protobuf这些编码格式减少消息体体积，提高存储量，减少序列化时间、提升性能。

Kafka Connect是一种基于Kafka的数据集成技术。Kafka Connect提供了出色的容错和可伸缩性，因为它作为分布式服务运行，并确保所有注册和配置的连接器始终运行。例如，即使集群中的某个Kafka Connect端点出现故障，剩余的Kafka Connect端点也会重新启动以前在现已终止的端点上运行的任何连接器，从而最大限度地减少停机时间并消除管理活动。Kafka Connect的任务分为“源”(source)和“汇”(sink)两种，“源”负责数据采集，“汇”负责数据投递。Kafka MirrorMaker 2.0和Debezium是两种典型的Kafka Connect应用。

Kafka MirrorMaker 2.0(MM2)旨在更轻松地将主题从一个Kafka群集镜像或复制到另一个群集。它使用Kafka Connect框架来简化配置和缩放。它动态检测主题的更改，并确保源和目标主题属性同步，包括偏移和分区。

CDC(Change Data Capture，变更数据捕获)是一种数据同步机制，它不断监控原始数据系统的更改，提取它们并将它们分发到上游系统，通过近乎实时地实现数据的增量加载，排除了批量数据加载的过程。

Debezium是一种CDC技术的开源实现。Debezium连接器通常通过将它们部署到Kafka Connect服务来操作，并配置一个或多个连接器来监视上游数据库并为它们在上游数据库中看到的所有变更生成数据变更事件。这些数据变更事件将写入Kafka，在那里它们可以被许多不同的应用程序独立使用。

流数据是指以大量、快速、时变的流形式持续到达的数据，流计算是指针对流数据的实时计算。

MQTT是一个轻量级的发布/订阅(publish/subscribe)形式的信息传输协议，设计原则是最低限度的网络带宽和设备资源的需求，同时确保可靠性和信息传递的可靠等级，设计思想是开放、简单、轻量、易于实现。该协议提供一对多的消息发布，解除应用程序耦合，使用TCP/IP提供网络连接，有三种消息发布服务质量，这些特点也使之成为设备到设备(M2M)、物联网(IoT)以及移动终端推送服务的理想协议。

公布号为CN112565333A，公开了一种基于kafka-connect的数据传递方法，描述了Kafka Connect的一种工作方式，但是并未描述Kafka Connect在工业企业中应用的具体场景。

公布号为CN108768826A，公开了基于MQTT和Kafka高并发场景下的消息路由方法，该方法充分利用MQTT轻量级协议在通信中支持数百万个设备同时连接的特点，引入Kafka集群以弥补MQTT协议不支持负载均衡的缺陷，并通过磁盘顺序写速度快的特点来应对高并发场景下的应用需求，大大提高了消息的传送速度，支持实时数据流的保存和异步处理。但该方法并不是基于Kafka Connec来实现。

发明内容

为克服现有技术的不足，本发明的目的是提供一种基于Kafka的工业企业实时泛在互联方法，Kafka技术生态，实现工业物联网数据和企业管理信息系统数据的融合，工业企业通常存在车间-产线-工厂-集团的多层级组织，建立信息逐层上报机制，实现云边端的全面协同。

为实现上述目的，本发明通过以下技术方案实现：

一种基于Kafka的工业企业实时泛在互联方法，具体包括以下内容：

1)工业物联网数据接入：基于MQTT协议接收工业物联网消息，通过数据集成工具Kafka Connect接收MQTT消息，并将消息写入Kafka队列中；

2)企业管理信息系统数据接入：基于CDC技术接收数据库变更信息，当数据库发生变更时，将变更信息写入到Kafka队列中；

3)信息逐层上报：将下层Kafka集群的Schema信息同步到上层Kafka SchemaRegistry中；将下层Kafka集群的数据信息和偏移量的信息同步到上层Kafka集群中；

4)数据融合：基于流计算技术进行数据融合，从而最终实现工业物联网数据和企业管理信息系统数据的融合。

工业物联网数据接入，具体包括以下步骤：

S11、在边缘侧或云端部署MQTT接收网关；

S12、在边缘侧或云端部署Kafka集群、Kafka Connect集群、启动ConfluentSchema Registry服务；

S13、在边缘侧部署边缘计算服务，通过OPC-UA、MODBUS或PROFIBUS工业协议，采集工业物联网信息，将消息Schema信息注册到Confluent Schema Registry中，将消息通过Avro格式进行编码，并通过MQTT协议向上传到MQTT网关中；

S14、配置Kafka Connect任务，通过MQTT的源连接器，读取MQTT中的工业物联网信息，并写入到Kafka的队列中。

企业管理信息系统数据接入，具体包括以下步骤：

S21、配置Debezium数据库，启用数据库binlog；

S22、配置Debezium数据库的Kafka Connect任务；

S23、Debezium数据库的Kafka任务监控上游数据库服务器，捕获配置的所有的数据库的变更，将变更通过Avro格式进行编码，并写入到Kafka的队列中。

信息逐层上报，具体包括以下步骤：

S31、实现下级Confluent Schema Registry到上级Confluent Schema Registry的复制合并功能；

S32、在上层Kafka集群中配置Kafka MirrorMaker 2，上层的Kafka MirrorMaker2通过拉取的方式实时同步下层Kafka集群信息；

S33、如果下层Kafka集群中使用包括Avro、ProtoBuf编码，则上层Kafka使用步骤S31中同步的Schema信息解析报文内容。

数据融合，具体包括以下步骤：

S41、使用流计算技术，根据业务场景实现工业物联网数据和企业管理信息系统数据的多流合并；

S42、将多流合并后的数据持久化到各类型数据库中。

与现有技术相比，本发明的有益效果是：

1.边缘计算服务通过Kafka Connect为Kafka提供了工业物联网数据；

2.企业管理信息系统通过Kafka Connect为Kafka提供了管理信息系统数据；

3.基于Kafka实现了多层级数据的逐层上报；

4.基于Schema Registry实现了模式信息共享。

附图说明

图1是基于Kafka的工业企业实时泛在互联方法的组成示意图。

具体实施方式

下面结合说明书附图对本发明进行详细地描述，但是应该指出本发明的实施不限于以下的实施方式。

见图1，一种基于Kafka的工业企业实时泛在互联方法，基于Kafka集群、KafkaConnect集群、Schema Registry、边缘计算服务、企业管理信息系统的软件系统实现，其中Kafka是核心组件，边缘计算服务通过Kafka Connect为Kafka提供工业物联网数据，企业管理信息系统通过Kafka Connect为Kafka提供管理信息系统数据，基于Kafka实现多层级数据的逐层上报，基于Schema Registry实现模式信息共享，最终实现工业企业的实时泛在互联；

具体包括以下内容：

1.工业物联网数据接入，具体包括以下步骤：

S11、在边缘侧或云端部署MQTT网关，例如EMQX；

S12、在边缘侧或云端部署Kafka集群、Kafka Connect集群、启动Schema Registry服务；

S13、在边缘侧部署边缘计算服务，通过OPC-UA、MODBUS、PROFIBUS的工业协议，采集工业物联网信息，将消息模式信息注册到Schema Registry中，将消息通过Avro格式进行编码，并通过MQTT协议上传到MQTT网关中；

S14、配置Kafka Connect任务，通过MQTT的源连接器，读取MQTT中的工业物联网信息，并写入到Kafka的队列中。

2.企业管理信息系统数据接入，具体包括以下步骤：

S21、配置数据库，启用数据库binlog；

S22、配置Debezium的Kafka Connect任务；

S23、Debezium的Kafka任务监控上游数据库服务器，捕获配置的所有的数据库的变更，将变更通过Avro格式进行编码，并写入到Kafka的队列中。

3.信息逐层上报：

工业企业通常存在车间-产线-工厂-集团的多层级组织，工业物联网、管理信息系统信息通常也是按层级来管理，因此要实现整个企业范围的信息融合，必须解决信息逐层传递的问题；

具体包括以下步骤：

S31、实现下级Schema Registry到上级Schema Registry的复制合并功能；

S32、在上层Kafka集群中配置Kafka MirrorMaker 2，上层的Kafka MirrorMaker2通过拉取的方式实时同步下层Kafka集群信息；

S33、如果下层Kafka集群中使用Avro、ProtoBuf编码，则上层Kafka使用步骤S31中同步的Schema信息解析报文内容。

4.数据融合，具体包括以下步骤：

S41、使用Kakfa KSQL或Flink实现流计算；

S42、将实时计算的数据持久化到各类型数据库中。

以下实施例在以本发明技术方案为前提下进行实施，给出了详细的实施方式和具体的操作过程，但本发明的保护范围不限于下述的实施例。下述实施例中所用方法如无特别说明均为常规方法。

【实施例1】

某大型工业企业要建设集团层级的统一的一体化平台，该集团下设多个生产基地，每个生产基地有若干生产厂，既要采集海量现场工业物联网实时数据，又要采集上百套既有企业管理信息系统的实时变更数据，实现了企业全域数据的跨业务实时查询和决策分析；具体方案为：

1.在每个生产厂部署边缘计算服务，边缘计算服务采集现场PLC设备信息，并通过MQTT协议上传到生产基地的私有云上。

2.生产基地的私有云上部署MQTT网关和Kafka集群，接收生产厂上传的PLC设备信息并写入生产基地私有云的Kafka集群中。

3.基地级管理信息系统的数据库上部署Debezium，采集基地级管理信息系统数据库实时变更信息，并写入生产基地私有云的Kafka集群中。

4.集团层级部署一套Kafka集群，并通过KafkaMM2从各生产基地同步模式信息，并实时拉取数据。

5.通过Flink实现对集团层级Kafka集群中数据的实时计算，并将结果写入到数据库中。

本发明的边缘计算服务通过Kafka Connect为Kafka提供了工业物联网数据；企业管理信息系统通过Kafka Connect为Kafka提供了管理信息系统数据；基于Kafka实现了多层级数据的逐层上报；基于Schema Registry实现了模式信息共享。