如何使用NiFi等构建IIoT系统

阅读原文时间：2023年07月11日阅读：1

您认为构建一个先进的工业物联网原型需要多长时间：

从传感器收集数据到每个工厂的网关

将传感器数据从一个或多个工厂移至云或数据中心

自动热部署新配置到所有边缘设备

支持大规模数据量和端到端安全性

使用正确的工具，您可以在不到一小时的时间内构建这样的系统！在此博客文章中，我将向您展示如何使用Raspberry Pi硬件和开源软件（MQTT代理、Apache NiFi、MiNiFi和MiNiFi C2 Server）实现高级IIoT原型。我将专注于体系结构，连接性，数据收集和自动重新配置。

工业物联网架构

有大量的物联网参考架构。通常，在工业环境中，您无法直接访问传感器和控制系统。网关用于桥接OT和IT世界。因此，IIoT体系结构通常包括边缘设备、网关、区域中心和最终的存储/处理系统。

下图显示了我们系统的全局体系结构以及将在每个级别使用的软件工具。

物联网架构

在边缘级别，传感器收集数字世界中的信息，并通过各种有线和无线协议（串口、RS-485、MODBUS、CAN总线、OPC UA、BLE、WiFi等）将其发送到网关。在我们的示例中，我们将使用各种传感器（光线、温度、摄像头、加速度计等），这些传感器通过WiFi将数据发送到网关。

网关是运行Mosquitto Broker和MiNiFi代理的Raspberry Pi。Mosquitto 是一个开源的轻量级消息传递代理，我们使用它通过MQTT协议公开传感器数据。MQTT的占地面积极小，使其适用于IoT应用程序和资源受限的硬件，例如电话或微控制器。

Apache MiNiFi是Apache NiFi的子项目，是一种轻量级代理，它实现了Apache NiFi的核心功能，侧重于边缘的数据收集。

MiNiFi的设计目标是：体积小，资源消耗低，代理的集中管理和边缘智能。通过站点到站点协议（S2S），可以将MiNiFi与NiFi轻松集成，以构建可扩展，安全并提供完整信息托管（来源）的端到端流管理解决方案。

在我们的系统中，MiNiFi将订阅Mosquitto Broker的所有主题，并将每条新消息转发到区域级别的NiFi。我们也可以使用它连接到SCADA系统或任何其他OT数据提供者。

在区域级别，我们有两个组成部分：

Apache NiFi 是一个功能强大的数据流平台，具有300多个现成的连接器。得益于其UI，设计数据流变得轻松快捷。

NiFi不会为了简单而放弃能力。确实，这是一个高度可扩展的分布式系统，具有可靠的交付、背压和负载分配。这些功能使NiFi成为物联网应用的绝佳工具，在这些应用中网络质量可能会面临挑战。

在我们的系统中，NiFi发挥着中心作用，即从每个工厂收集数据并将其路由到多个系统和应用程序（HDFS、HBase、Kafka、S3等）。

MiNiFi C2服务器（MiNiFi Command＆Control）是当前正在开发的Apache NiFi的另一个子项目。它的作用是为野外成百上千的MiNiFi代理提供配置的中心点。C2服务器管理应用程序的版本化类（MiNiFi流配置），并通过Rest API公开它们。MiNiFi代理可以以定义的频率连接到该API以更新其配置。

一旦数据到达公司服务器、云或数据中心，便可以实施大量应用程序。实时监控，过程分析和优化或预测性维护就是其中的几个示例。数据处理和用例实现将在以后的文章中讨论。

系统实施

让我们开始构建我们的原型。

准备Raspberry Pi：MQTT和MiNiFi

要安装Mosquitto MQTT代理和MiNiFi代理，请在Raspberry Pi上运行以下命令。

为了减小体积，MiNiFi打包了最少的默认处理器集。通过在lib目录中部署NAR（NiFi存档），可以添加任何NiFi处理器。在下面的块的最后一个命令中，我添加了MQTT处理器的NAR。

sudo apt-get update#install and run Mosquitto broker on default port 1883sudo apt-get install mosquitomosquito#install and prepare MiNiFi agentwget http://apache.crihan.fr/dist/nifi/minifi/0.4.0/minifi-0.4.0-bin.tar.gz tar -xvf minifi-0.4.0-bin.tar.gzcd minifi-0.4.0#add mqtt processorwget https://github.com/ahadjidj-hw/NiFi/raw/master/nifi-mqtt-nar-1.5.0.nar -P ./lib/

默认情况下，配置MiNiFi代理程序需要编辑文件./conf/config.yml以包括使用的处理器及其配置的列表。可以手动编写配置，也可以使用NiFi UI设计配置，然后将流程导出为模板。该模板是一个XML文件，我们需要使用MiNiFi工具包将其转换为YML文件。这是一个配置文件的示例，该文件尾部一个文件，并通过S2S将每一行发送到远程NiFi。

对于我们的项目，我们将不使用这些手动步骤。由于许多MiNiFi代理在地理位置分散的工厂上运行，因此无法手动停止、编辑config.yml，然后在每次需要更改其配置时重新启动每个代理。

MiNiFi使用“更改接收器”，通过该“更改接收器”将可能的新配置通知给代理。变更接收器是可插入模块，目前支持三种OOTB接收器：

FileChangeIngestor

RestChangeIngestor

PullHttpChangeIngestor

我们将每隔一段时间使用PullHttpChangeIngestor来查询C2服务器并下载任何可用的新配置。要配置此摄取器，请编辑文件./conf/bootstrap.conf，取消注释相应的行，并按如下所示设置摄取器属性：

nifi.minifi.notifier.ingestors=org.apache.nifi.minifi.bootstrap.configuration.ingestors.PullHttpChangeIngestor# Hostname on which to pull configurations fromnifi.minifi.notifier.ingestors.pull.http.hostname=c2-server# Port on which to pull configurations fromnifi.minifi.notifier.ingestors.pull.http.port=10080# Path to pull configurations fromnifi.minifi.notifier.ingestors.pull.http.path=/c2/config# Query string to pull configurations withnifi.minifi.notifier.ingestors.pull.http.query=class=iot-minifi-raspberry-agent# Period on which to pull configurations from, defaults to 5 minutes if commented outnifi.minifi.notifier.ingestors.pull.http.period.ms=60000

使用此配置，每个MiNiFi代理每隔1分钟会在http://c2-server:10080/c2/config 查询C2服务器REST API，并要求“ iot-minifi-raspberry-agent”类的最新配置。

注意：1分钟的频率仅用于演示目的。您不会如此频繁地更新代理。

现在就不要启动您的代理，让我们转到区域级别并配置MiNiFi C2服务器和NiFi。

安装和配置MiNiFi C2服务器

在可从MiNiFi代理访问的公共服务器上安装MiNiFi C2服务器。您可以将C2分层部署用于受网络限制的应用程序，如以下几行所述。运行以下命令以安装C2服务器：

wget http://apache.crihan.fr/dist/nifi/minifi/0.4.0/minifi-c2-0.4.0-bin.tar.gztar -xvf minifi-c2-0.4.0-bin.tar.gzcd minifi-c2-0.4.0

C2服务器通过按类组织的REST API公开MiNiFi应用程序。C2支持可插拔的“配置提供程序”，当前支持：

该CacheConfigurationProvider，它着眼于目录上的文件系统或S3

该DelegatingConfigurationProvider，委托给另一C2服务器允许分层C2结构

该NiFiRestConfigurationProvider，从一个NiFi实例在其REST API拉模板

配置C2服务器以使用NiFi作为配置提供程序。编辑文件./conf/minifi-c2-context.xml 并提供NiFi服务器地址 http://nifi-dev:8080

安装和配置NiFi服务器

在C2服务器可访问的服务器上安装NiFi并运行它。

wget http://apache.crihan.fr/dist/nifi/1.6.0/nifi-1.6.0-bin.tar.gz tar -xvf nifi-1.6.0-bin.tar.gzcd nifi-1.6.0./bin/nifi.sh start

让我们在http://nifi-dev:8080/nifi/ 上连接到NiFi UI，并创建将在MiNiFi代理中运行的流。但是在此之前，请在根画布上添加一个输入端口，并将其命名为“来自Raspberry MiNiFi”。NiFi将从此处接收来自MiNiFi的流文件。

添加consumerMQTT处理器以订阅Mosquitto代理并订阅iot / sensors下的所有主题。请注意，此处的tcp://raspberrypi:1883与tcp://localhost:1883等效，因为此流将在Raspberry Pi上运行。

使用UpdateAttribute处理器添加“版本”属性，我们将使用该属性来显示重新配置功能。您可以添加所需的任何属性：时间戳记，座席名称，位置等。

最后，添加一个远程进程组（RPG）以将使用的事件发送到NiFi。连接这三个处理器。

现在，您的流程类似于以下屏幕截图。左侧的数据流将在NiFi中运行，以接收来自MiNiFi的数据。正确的流程仅用于设计，并且可以在每个Raspberry Pi上有效运行。

将正确的流程另存为模板，名称为“ iot-minifi-raspberry-agent.v1”。这里的命名约定非常重要。我们必须使用与MiNiFi引导程序配置中使用的类名相同的名称。

部署并启动应用程序

在Raspberry Pi上启动MiNiFi代理之前，让我们看一下C2服务器是否配置正确。在Web浏览器中打开以下URL：http://c2-server:10080/c2/config?class=iot-minifi-raspberry-agent&version=1 。C2 Server答复一个文件，其中包含我们构建的模板的配置，格式为YML。那很棒。

C2 Rest API调用的结果

如果查看C2日志，则可以看到服务器收到带有参数{class = [iot-minifi-raspberry-agent]，version = [1]}的查询。

REST API调用后的C2服务器日志

现在，体系结构的不同组件（MQTT，MiNiFi，NiFi和C2）之间的通信正在工作，请使用以下命令在Raspberry Pi上启动MiNiFi代理：

./bin/minifi.sh start

几秒钟后，您会看到以下C2服务器日志。主机192.168.1.50（这是Raspberry Pi的IP地址）要求C2服务器为其提供最新版本的类“ iot-minifi-raspberry-agent”。与我们先前使用Web浏览器进行的呼叫相比，您会注意到MiNiFi代理未指定版本。如果现在在./conf/config.yml中打开MiNiFi代理配置，您将找到我们从C2 Rest API中检索到的相同conf文件。

C2服务器日志

此外，MQTT还显示MiNiFi代理已连接到代理并订阅了主题iot / sensors /＃

MiNiFi代理启动后的MQTT日志

完善！IIoT系统运行得像灵符。现在，让我们启动传感器以生成数据并将其发布在MQTT中。然后，MiNiFi将开始使用数据并将其发送到NiFi，如以下屏幕截图所示，其中我们已收到196条消息。

现在，让我们使用NiFi的来源功能检查这些消息之一。该数据来自光传感器“ iot/sensors/LightIntensity/z”，应用程序版本为1。

自动化热重新部署

现在我们的IIoT正在运行，并且数据正在从每个工厂流到我们的数据中心，让我们部署一个新的应用程序。对于我们的测试，我们将对我们的MiNiFi代理配置进行较小的修改。转到NiFi网络用户界面，然后编辑updateAttribute处理器。将“版本”属性设置为2而不是1，并将流保存在新模板“ iot-minifi-raspberry-agent.v2”中。就这样！新的应用程序将自动部署。

您可以在下面看到C2服务器日志，显示检测到新版本V2。C2服务器的缓存中没有此版本，因此开始下载和转换过程。