Apache NiFi - 快速指南

---

---

Apache NiFi - 简介

Apache NiFi 是一个功能强大、易于使用且可靠的系统，用于在不同的系统之间处理和分发数据。它基于 NSA 开发的 Niagara Files 技术，并在 8 年后捐赠给 Apache 软件基金会。它根据 Apache 许可证版本 2.0（2004 年 1 月）分发。Apache NiFi 的最新版本为 1.7.1。

Apache NiFi 是一个实时数据摄取平台，可以传输和管理不同源系统和目标系统之间的数据传输。它支持各种各样的数据格式，例如日志、地理位置数据、社交信息流等。它还支持许多协议，例如 SFTP、HDFS 和 KAFKA 等。对各种数据源和协议的支持使得该平台在许多 IT 组织中广受欢迎。

Apache NiFi - 一般特性

Apache NiFi 的一般特性如下：

• Apache NiFi 提供了一个基于 Web 的用户界面，在设计、控制、反馈和监控之间提供了无缝的体验。

• 它具有高度可配置性。这有助于用户实现保证交付、低延迟、高吞吐量、动态优先级、背压以及在运行时修改数据流。

• 它还提供数据溯源模块来跟踪和监控从数据流开始到结束的数据。

• 开发人员可以根据自己的需求创建自己的自定义处理器和报告任务。

• NiFi 还支持安全的协议，如 SSL、HTTPS、SSH 和其他加密方式。

• 它还支持用户和角色管理，并且可以与 LDAP 配置以进行授权。

Apache NiFi - 关键概念

Apache NiFi 的关键概念如下：

• 处理组 - 它是一组 NiFi 数据流，帮助用户以分层的方式管理和维护数据流。

• 数据流 - 通过连接不同的处理器来创建，以便根据需要传输和修改数据，从一个或多个数据源到另一个目标数据源。

• 处理器 - 处理器是一个 Java 模块，负责从源系统获取数据或将其存储在目标系统中。其他处理器还用于在 FlowFile 中添加属性或更改内容。

• FlowFile - 它是 NiFi 的基本用法，表示从源系统中提取到 NiFi 的单个数据对象。NiFi 处理器在 FlowFile 从源处理器移动到目标处理器的过程中对其进行更改。不同的事件（如创建、克隆、接收等）由数据流中的不同处理器在 FlowFile 上执行。

• 事件 - 事件表示 FlowFile 在遍历 NiFi 数据流时发生的更改。这些事件在数据溯源中进行跟踪。

• 数据溯源 - 它是一个存储库。它还有一个 UI，允许用户检查有关 FlowFile 的信息，并在处理 FlowFile 期间出现任何问题时帮助进行故障排除。

Apache NiFi 优点

• Apache NiFi 通过使用 SFTP 从远程机器获取数据，并保证数据血缘关系。

• Apache NiFi 支持集群，因此它可以在多个节点上运行，每个节点都使用相同的数据流处理不同的数据，从而提高数据处理性能。

• 它还提供用户级别、处理组级别和其他模块的安全策略。

• 其 UI 也可以在 HTTPS 上运行，这使得用户与 NiFi 的交互更加安全。

• NiFi 支持大约 188 个处理器，用户还可以创建自定义插件以支持各种各样的数据系统。

Apache NiFi 缺点

• 当节点在用户对其进行任何更改时与 NiFi 集群断开连接时，flow.xml 将变得无效。节点无法重新连接到集群，除非管理员手动从连接的节点复制 flow.xml。

• 在主节点切换的情况下，Apache NiFi 存在状态持久性问题，这有时会导致处理器无法从源系统获取数据。

Apache NiFi - 基本概念

Apache NiFi 由一个 Web 服务器、一个数据流控制器和一个在 Java 虚拟机上运行的处理器组成。它还有 3 个存储库：FlowFile 存储库、内容存储库和溯源存储库，如下所示。

FlowFile 存储库

此存储库存储通过 Apache NiFi 数据流的每个 FlowFile 的当前状态和属性。此存储库的默认位置在 Apache NiFi 的根目录中。可以通过更改名为“nifi.flowfile.repository.directory”的属性来更改此存储库的位置。

内容存储库

此存储库包含 NiFi 中所有 FlowFile 中存在的所有内容。其默认目录也在 NiFi 的根目录中，可以使用“org.apache.nifi.controller.repository.FileSystemRepository”属性更改它。此目录使用磁盘中的大量空间，因此建议在安装磁盘中有足够的可用空间。

溯源存储库

该存储库跟踪并存储在 NiFi 中流动的所有 FlowFile 的所有事件。有两个溯源存储库：**易失性溯源存储库**（在此存储库中，所有溯源数据在重新启动后都会丢失）和**持久性溯源存储库**。其默认目录也在 NiFi 的根目录中，可以使用“org.apache.nifi.provenance.PersistentProvenanceRepository”和“org.apache.nifi.provenance.VolatileProvenanceRepositor”属性分别为相应的存储库更改。

Apache NiFi - 环境设置

在本章中，我们将学习 Apache NiFi 的环境设置。Apache NiFi 的安装步骤如下：

步骤 1 - 在您的计算机上安装当前版本的 Java。请在您的机器上设置 JAVA_HOME。您可以检查如下所示的版本

在 Windows 操作系统 (OS) 中（使用命令提示符）：

> java -version

在 UNIX OS 中（使用终端）：

$ echo $JAVA_HOME

步骤 2 - 从 https://nifi.apache.org/download.html 下载 Apache NiFi

• 对于 Windows OS，下载 ZIP 文件。

• 对于 UNIX OS，下载 TAR 文件。

• 对于 Docker 镜像，请访问以下链接 https://hub.docker.com/r/apache/nifi/.

步骤 3 - Apache NiFi 的安装过程非常简单。过程因 OS 而异：

• Windows OS - 解压缩 zip 包，Apache NiFi 就安装好了。

• UNIX OS - 将 tar 文件解压缩到任何位置，Logstash 就安装好了。

$tar -xvf nifi-1.6.0-bin.tar.gz

步骤 4 - 打开命令提示符，转到 NiFi 的 bin 目录。例如，C:\nifi-1.7.1\bin，并执行 run-nifi.bat 文件。

C:\nifi-1.7.1\bin>run-nifi.bat

步骤 5 - NiFi UI 启动需要几分钟。用户可以检查 nifi-app.log，一旦 NiFi UI 启动，用户就可以输入 https://:8080/nifi/ 来访问 UI。

Apache NiFi - 用户界面

Apache 是一个基于 Web 的平台，用户可以使用 Web UI 访问。NiFi UI 非常交互式，并提供了大量有关 NiFi 的信息。如下图所示，用户可以访问以下属性的信息：

• 活动线程
• 总排队数据
• 正在传输的远程处理组
• 未传输的远程处理组
• 正在运行的组件
• 已停止的组件
• 无效的组件
• 已禁用的组件
• 最新的版本化处理组
• 本地修改的版本化处理组
• 过时的版本化处理组
• 本地修改且过时的版本化处理组
• 同步失败的版本化处理组

Apache NiFi 的组件

Apache NiFi UI 包含以下组件：

处理器

用户可以将处理器图标拖到画布上，并为 NiFi 中的数据流选择所需的处理器。

输入端口

以下图标被拖到画布上以将输入端口添加到任何数据流中。

输入端口用于从不在该处理组中的处理器获取数据。

拖动此图标后，NiFi 会要求输入输入端口的名称，然后将其添加到 NiFi 画布。

输出端口

以下图标被拖到画布上以将输出端口添加到任何数据流中。

输出端口用于将数据传输到不在该处理组中的处理器。

拖动此图标后，NiFi 会要求输入输出端口的名称，然后将其添加到 NiFi 画布。

处理组

用户使用以下图标在 NiFi 画布中添加处理组。

拖动此图标后，NiFi 会要求输入处理组的名称，然后将其添加到 NiFi 画布。

远程处理组

这用于在 NiFi 画布中添加远程处理组。

漏斗

漏斗用于将处理器的输出传输到多个处理器。用户可以使用以下图标在 NiFi 数据流中添加漏斗。

模板

此图标用于将数据流模板添加到 NiFi 画布。这有助于在相同或不同的 NiFi 实例中重用数据流。

拖动后，用户可以选择已添加到 NiFi 中的模板。

标签

这些用于在 NiFi 画布上添加有关 NiFi 中任何组件的文本。它提供了一系列用户可用于添加美观效果的颜色。

Apache NiFi - 处理器

Apache NiFi 处理器是创建数据流的基本模块。每个处理器都有不同的功能，这有助于创建输出 FlowFile。下图所示的数据流使用 GetFile 处理器从一个目录中获取文件，并使用 PutFile 处理器将其存储到另一个目录中。

GetFile

GetFile 处理器用于从特定目录中获取特定格式的文件。它还为用户提供了其他选项以更好地控制获取过程。我们将在下面的属性部分讨论它。

GetFile 设置

以下是 GetFile 处理器的不同设置：

名称

在“名称”设置中，用户可以为处理器定义任何名称，可以根据项目或使其名称更有意义。

启用

用户可以使用此设置启用或禁用处理器。

惩罚持续时间

此设置允许用户在 FlowFile 失败时添加惩罚时间持续时间。

产出持续时间

此设置用于指定处理器的产出时间。在此持续时间内，进程不会再次调度。

公告级别

此设置用于指定该处理器的日志级别。

自动终止关系

此设置包含该特定流程所有可用关系的复选框列表。通过选中复选框，用户可以对处理器进行编程，使其在该事件发生时终止 FlowFile，并且不将其进一步发送到流程中。

GetFile 调度

以下是 GetFile 处理器提供的调度选项：

调度策略

您可以通过选择“时间驱动”来按时间计划流程，或者通过选择“CRON 驱动”选项来指定 CRON 字符串。

并发任务

此选项用于定义此处理器的并发任务调度。

执行

用户可以使用此选项定义是在所有节点上运行处理器还是仅在主节点上运行。

运行计划

它用于定义时间驱动策略的时间或 CRON 驱动策略的 CRON 表达式。

GetFile 属性

GetFile 提供了多个属性，如下面的图像所示，范围从必填属性（如输入目录和文件过滤器）到可选属性（如路径过滤器和最大文件大小）。用户可以使用这些属性来管理文件获取过程。

GetFile 注释

此部分用于指定有关处理器的任何信息。

PutFile

PutFile 处理器用于将文件从数据流存储到特定位置。

PutFile 设置

PutFile 处理器具有以下设置：

名称

在“名称”设置中，用户可以根据项目或使其名称更有意义的方式为处理器定义任何名称。

启用

用户可以使用此设置启用或禁用处理器。

惩罚持续时间

此设置允许用户在 FlowFile 失败的情况下添加罚时持续时间。

产出持续时间

此设置用于指定处理器的产出时间。在此持续时间内，流程不会再次调度。

公告级别

此设置用于指定该处理器的日志级别。

自动终止关系

此设置包含该特定流程所有可用关系的复选框列表。通过选中复选框，用户可以对处理器进行编程，使其在该事件发生时终止 FlowFile，并且不将其进一步发送到流程中。

PutFile 调度

以下是 PutFile 处理器提供的调度选项：

调度策略

您可以通过选择“定时器驱动”或选择“CRON 驱动”选项来指定 CRON 字符串，从而按时间计划流程。还有一个实验性策略“事件驱动”，它将在特定事件发生时触发处理器。

并发任务

此选项用于定义此处理器的并发任务调度。

执行

用户可以使用此选项定义是在所有节点上运行处理器还是仅在主节点上运行。

运行计划

它用于定义定时器驱动策略的时间或 CRON 驱动策略的 CRON 表达式。

PutFile 属性

PutFile 处理器提供了一些属性，例如“目录”，用于指定文件传输的输出目录，以及其他用于管理传输的属性，如下面的图像所示。

PutFile 注释

此部分用于指定有关处理器的任何信息。

Apache NiFi - 处理器分类

在本章中，我们将讨论 Apache NiFi 中的流程分类。

数据摄取处理器

数据摄取类别下的处理器用于将数据摄取到 NiFi 数据流中。这些通常是 Apache NiFi 中任何数据流的起点。属于这些类别的一些处理器包括 GetFile、GetHTTP、GetFTP、GetKAFKA 等。

路由和中介处理器

路由和中介处理器用于根据这些 FlowFile 的属性或内容中的信息将 FlowFile 路由到不同的处理器或数据流。这些处理器还负责控制 NiFi 数据流。属于此类别的一些处理器包括 RouteOnAttribute、RouteOnContent、ControlRate、RouteText 等。

数据库访问处理器

此数据库访问类别的处理器能够从数据库中选择或插入数据，或执行和准备其他 SQL 语句。这些处理器主要使用 Apache NiFi 的数据连接池控制器设置。属于此类别的一些处理器包括 ExecuteSQL、PutSQL、PutDatabaseRecord、ListDatabaseTables 等。

属性提取处理器

属性提取处理器负责在 NiFi 数据流中提取、分析和更改 FlowFile 属性处理。属于此类别的一些处理器包括 UpdateAttribute、EvaluateJSONPath、ExtractText、AttributesToJSON 等。

系统交互处理器

系统交互处理器用于在任何操作系统中运行进程或命令。这些处理器还以多种语言运行脚本，以与各种系统交互。属于此类别的一些处理器包括 ExecuteScript、ExecuteProcess、ExecuteGroovyScript、ExecuteStreamCommand 等。

数据转换处理器

属于数据转换的处理器能够更改 FlowFile 的内容。这些可以用于完全替换 FlowFile 的数据，通常在用户必须将 FlowFile 作为 HTTP 主体发送以调用 InvokeHTTP 处理器时使用。属于此类别的一些处理器包括 ReplaceText、JoltTransformJSON 等。

发送数据处理器

发送数据处理器通常是数据流中的最终处理器。这些处理器负责将数据存储或发送到目标服务器。在成功存储或发送数据后，这些处理器将 FlowFile 丢弃，并使用成功关系。属于此类别的一些处理器包括 PutEmail、PutKafka、PutSFTP、PutFile、PutFTP 等。

拆分和聚合处理器

这些处理器用于拆分和合并 FlowFile 中存在的内容。属于此类别的一些处理器包括 SplitText、SplitJson、SplitXml、MergeContent、SplitContent 等。

HTTP 处理器

这些处理器处理 HTTP 和 HTTPS 调用。属于此类别的一些处理器包括 InvokeHTTP、PostHTTP、ListenHTTP 等。

AWS 处理器

AWS 处理器负责与 Amazon Web Services 系统交互。属于此类别的一些处理器包括 GetSQS、PutSNS、PutS3Object、FetchS3Object 等。

Apache NiFi - 处理器关系

在 Apache NiFi 数据流中，FlowFile 通过连接从一个处理器移动到另一个处理器，这些连接使用处理器之间的关系进行验证。每当创建连接时，开发人员都会选择这些处理器之间的一种或多种关系。

如上图所示，黑色矩形中的复选框是关系。如果开发人员选中这些复选框，则当关系成功或失败或两者兼有时，FlowFile 将在该特定处理器中终止。

成功

当处理器成功处理 FlowFile（例如从任何数据源存储或获取数据，而不会遇到任何连接、身份验证或任何其他错误）时，FlowFile 将进入成功关系。

失败

当处理器无法在没有错误的情况下处理 FlowFile（例如身份验证错误或连接问题等）时，FlowFile 将进入失败关系。

开发人员还可以使用连接将 FlowFile 传输到其他处理器。开发人员可以选择并对其进行负载平衡，但负载平衡仅在 1.8 版中发布，本教程中不会介绍。

如上图所示，红色标记的连接具有失败关系，这意味着所有带有错误的 FlowFile 将进入左侧的处理器，并且所有没有错误的 FlowFile 将分别传输到绿色标记的连接。

现在让我们继续讨论其他关系。

comms.failure

当由于通信故障无法从远程服务器获取 FlowFile 时，将满足此关系。

not.found

对于我们从远程服务器收到“未找到”消息的任何 FlowFile，都将移动到**not.found**关系。

permission.denied

当 NiFi 由于权限不足而无法从远程服务器获取 FlowFile 时，它将通过此关系移动。

Apache NiFi - FlowFile

FlowFile 是 Apache NiFi 中的基本处理实体。它包含数据内容和属性，NiFi 处理器使用这些内容和属性来处理数据。文件内容通常包含从源系统获取的数据。Apache NiFi FlowFile 的最常见属性包括：

UUID

它代表通用唯一标识符，是 NiFi 生成的 FlowFile 的唯一标识。

文件名

此属性包含该 FlowFile 的文件名，并且不应包含任何目录结构。

文件大小

它包含 Apache NiFi FlowFile 的大小。

mime.type

它指定此 FlowFile 的 MIME 类型。

path

此属性包含 FlowFile 所属文件的相对路径，并且不包含文件名。

Apache NiFi - 队列

Apache NiFi 数据流连接具有一个排队系统来处理大量数据流入。这些队列可以处理非常大量的 FlowFile，以允许处理器按顺序处理它们。

上图中的队列具有 1 个通过成功关系传输的 FlowFile。用户可以通过选择下拉列表中的“列出队列”选项来检查 FlowFile。在任何过载或错误的情况下，用户还可以通过选择“清空队列”选项来清除队列，然后用户可以重新启动流程以再次获取数据流中的这些文件。

队列中的 FlowFile 列表包含位置、UUID、文件名、文件大小、队列持续时间和血缘持续时间。用户可以通过单击 FlowFile 列表第一列中存在的 info 图标来查看 FlowFile 的所有属性和内容。

Apache NiFi - 处理组

在 Apache NiFi 中，用户可以在不同的流程组中维护不同的数据流。这些组可以基于 Apache NiFi 实例支持的不同项目或组织。

NiFi UI 顶部菜单中的第四个符号（如上图所示）用于在 NiFi 画布中添加流程组。“Tutorialspoint.com_ProcessGroup”流程组包含一个数据流，其中目前有四个处理器处于停止状态，如上图所示。可以以分层方式创建流程组，以便以更易于理解的结构管理数据流。

在 NiFi UI 的页脚中，您可以看到流程组，并且可以返回到用户当前所在的流程组的顶部。

要查看 NiFi 中存在的流程组的完整列表，用户可以使用 NiFi UI 左侧顶部存在的菜单转到摘要。在摘要中，有一个流程组选项卡，其中列出了所有流程组，以及版本状态、传输/大小、输入/大小、读取/写入、输出/大小等参数，如下面的图片所示。

Apache NiFi - 标签

Apache NiFi 提供标签，使开发人员能够编写有关 NiFi 画布中存在的组件的信息。NiFi UI 顶部菜单中最左边的图标用于在 NiFi 画布中添加标签。

开发人员可以通过右键单击标签并从菜单中选择适当的选项来更改标签的颜色和文本大小。

Apache NiFi - 配置

Apache NiFi 是一个高度可配置的平台。conf 目录中的 nifi.properties 文件

包含大部分配置。

Apache NiFi 的常用属性如下：

核心属性

此部分包含运行 NiFi 实例所必需的属性。

状态管理

这些属性用于存储组件的状态，有助于启动处理，其中组件在重新启动后以及在下一个计划运行中保留。

FlowFile 存储库

现在让我们了解 FlowFile 存储库的重要细节 -

Apache NiFi - 管理

Apache NiFi 提供对 ambari、zookeeper 等多种工具的支持，用于管理目的。NiFi 还在 nifi.properties 文件中提供配置，以便为管理员设置 HTTPS 和其他内容。

zookeeper

NiFi 本身不处理集群中的投票过程。这意味着当创建集群时，所有节点都是主节点和协调器。因此，配置 ZooKeeper 以管理主节点和协调器的投票。nifi.properties 文件包含一些用于设置 ZooKeeper 的属性。

启用 HTTPS

要通过 HTTPS 使用 NiFi，管理员必须生成密钥库和信任库，并在 nifi.properties 文件中设置一些属性。TLS 工具包可用于生成启用 Apache NiFi 中 HTTPS 所需的所有密钥。

其他管理属性

还有一些其他属性，管理员使用这些属性来管理 NiFi 及其服务连续性。

Apache NiFi - 创建数据流

Apache NiFi 提供大量组件，以帮助开发人员为任何类型的协议或数据源创建数据流。要创建流，开发人员会将组件从菜单栏拖动到画布上，然后通过单击并从一个组件拖动到另一个组件来连接它们。

通常，NiFi 在流的开头有一个侦听器组件，例如 getfile，它从源系统获取数据。在另一端有一个发射器组件，例如 putfile，并且中间有一些组件来处理数据。

例如，让我们创建一个流，它从一个目录获取一个空文件，在该文件中添加一些文本，然后将其放入另一个目录。

• 首先，将处理器图标拖到 NiFi 画布上，然后从列表中选择 GetFile 处理器。

• 创建一个输入目录，例如 **c:\inputdir**。

• 右键单击处理器并选择配置，在属性选项卡中添加 **输入目录 (c:\inputdir)** 并单击应用，然后返回画布。

• 将处理器图标拖到画布上，然后从列表中选择 ReplaceText 处理器。

• 右键单击处理器并选择配置。在**属性**选项卡中，在替换值的文本框中添加一些文本，例如**“Hello tutorialspoint.com”**，然后单击应用。

• 转到设置选项卡，选中右侧的失败复选框，然后返回画布。

• 将 GetFIle 处理器连接到 ReplaceText 的成功关系。

• 将处理器图标拖到画布上，然后从列表中选择 PutFile 处理器。

• 创建一个输出目录，例如 **c:\outputdir**。

• 右键单击处理器并选择配置。在属性选项卡中，添加 **目录 (c:\outputdir)** 并单击应用，然后返回画布。

• 转到设置选项卡，选中右侧的失败和成功复选框，然后返回画布。

• 将 ReplaceText 处理器连接到 PutFile 的成功关系。

• 现在启动流并在输入目录中添加一个空文件，您将看到它将移动到输出目录，并且文本将添加到该文件中。

通过遵循上述步骤，开发人员可以选择任何处理器和其他 NiFi 组件来为其组织或客户创建合适的流。

Apache NiFi - 模板

Apache NiFi 提供了模板的概念，这使得更容易重用和分发 NiFi 流。其他开发人员或其他 NiFi 集群可以使用这些流。它还有助于 NiFi 开发人员在 GitHub 等存储库中共享他们的工作。

创建模板

让我们为我们在第 15 章“Apache NiFi - 创建流”中创建的流创建一个模板。

使用 Shift 键选择流的所有组件，然后单击 NiFi 画布左侧的创建模板图标。您还可以看到一个工具箱，如上图所示。单击上面图片中蓝色标记的**创建模板**图标。输入模板的名称。开发人员还可以添加描述，这是可选的。

下载模板

然后转到 NiFi UI 右上角菜单中的 NiFi 模板选项，如下面的图片所示。

现在单击要下载的模板的下载图标（位于列表右侧）。将下载一个带有模板名称的 XML 文件。

上传模板

要在 NiFi 中使用模板，开发人员必须使用 UI 将其 xml 文件上传到 NiFi。在创建模板图标旁边有一个上传模板图标（在下面的图片中用蓝色标记），单击该图标并浏览 xml。

添加模板

在 NiFi UI 的顶部工具栏中，模板图标位于标签图标之前。该图标在下面的图片中用蓝色标记。

拖动模板图标并从下拉列表中选择模板，然后单击添加。它会将模板添加到 NiFi 画布。

Apache NiFi - API

NiFi 提供大量 API，帮助开发人员从任何其他工具或自定义开发的应用程序中更改和获取 NiFi 的信息。在本教程中，我们将使用 Google Chrome 中的 postman 应用程序来解释一些示例。

要将 postman 添加到您的 Google Chrome，请访问下面提到的 URL 并单击添加到 Chrome 按钮。您现在将看到一个新应用添加到您的 Google Chrome。

chrome 网上应用商店

NiFi rest API 的当前版本是 1.8.0，文档位于下面提到的 URL 中。

https://nifi.apache.org/docs/nifi-docs/rest-api/index.html

以下是最常用的 NiFi rest API 模块 -

• http://<nifi url>:<nifi port>/nifi-api/<api-path>

• 如果启用了 HTTPS，则为 https://<nifi url>:<nifi port>/nifi-api/<api-path>

现在让我们考虑一个示例并在 postman 上运行以获取有关正在运行的 NiFi 实例的详细信息。

请求

GET https://:8080/nifi-api/flow/about

响应

{
   "about": {
      "title": "NiFi",
      "version": "1.7.1",
      "uri": "https://:8080/nifi-api/",
      "contentViewerUrl": "../nifi-content-viewer/",
      "timezone": "SGT",
      "buildTag": "nifi-1.7.1-RC1",
      "buildTimestamp": "07/12/2018 12:54:43 SGT"
   }
}

Apache NiFi - 数据溯源

Apache NiFi 记录并存储数据流中所有摄取数据事件的信息。数据溯源存储库存储这些信息，并提供 UI 用于搜索这些事件信息。数据溯源可以访问 NiFi 全局级别和处理器级别。

下表列出了 NiFi 数据溯源事件列表中的不同字段：

要获取有关事件的更多信息，用户可以单击 NiFi 数据溯源 UI 第一列中显示的信息图标。

nifi.properties 文件中有一些属性用于管理 NiFi 数据溯源存储库。

Apache NiFi - 监控

在 Apache NiFi 中，有多种方法可以监控系统的不同统计信息，例如错误、内存使用情况、CPU 使用情况、数据流统计信息等。本教程将讨论其中最常用的几种。

内置监控

在本节中，我们将详细了解 Apache NiFi 中的内置监控。

公告板

公告板实时显示 NiFi 处理器生成的最新错误和警告。要访问公告板，用户需要转到右侧的向下菜单并选择“公告板”选项。它会自动刷新，用户也可以禁用它。用户还可以通过双击错误导航到实际的处理器。用户还可以通过以下方式过滤公告：

• 按消息
• 按名称
• 按 ID
• 按组 ID

数据溯源 UI

要监控任何特定处理器或整个 NiFi 中发生的事件，用户可以从与公告板相同的菜单访问数据溯源。用户还可以通过以下字段过滤数据溯源存储库中的事件：

• 按组件名称
• 按组件类型
• 按类型

NiFi 摘要 UI

Apache NiFi 摘要也可以从与公告板相同的菜单访问。此 UI 包含有关该特定 NiFi 实例或集群的所有组件的信息。它们可以通过名称、类型或 URI 进行过滤。不同组件类型有不同的选项卡。以下是可以监控的 NiFi 摘要 UI 中的组件：

• 处理器
• 输入端口
• 输出端口
• 远程进程组
• 连接
• 进程组

在此 UI 中，在右下角有一个名为“系统诊断”的链接，用于检查 JVM 统计信息。

报告任务

Apache NiFi 提供多个报告任务来支持外部监控系统，例如 Ambari、Grafana 等。开发人员可以创建自定义报告任务，也可以配置内置任务将 NiFi 的指标发送到外部监控系统。下表列出了 NiFi 1.7.1 提供的报告任务。

NiFi API

有一个名为“系统诊断”的 API，可用于在任何自定义开发的应用程序中监控 NiFi 统计信息。让我们在 Postman 中检查此 API。

请求

https://:8080/nifi-api/system-diagnostics

响应

{
   "systemDiagnostics": {
      "aggregateSnapshot": {
         "totalNonHeap": "183.89 MB",
         "totalNonHeapBytes": 192819200,
         "usedNonHeap": "173.47 MB",
         "usedNonHeapBytes": 181894560,
         "freeNonHeap": "10.42 MB",
         "freeNonHeapBytes": 10924640,
         "maxNonHeap": "-1 bytes",
         "maxNonHeapBytes": -1,
         "totalHeap": "512 MB",
         "totalHeapBytes": 536870912,
         "usedHeap": "273.37 MB",
         "usedHeapBytes": 286652264,
         "freeHeap": "238.63 MB",
         "freeHeapBytes": 250218648,
         "maxHeap": "512 MB",
         "maxHeapBytes": 536870912,
         "heapUtilization": "53.0%",
         "availableProcessors": 4,
         "processorLoadAverage": -1,
         "totalThreads": 71,
         "daemonThreads": 31,
         "uptime": "17:30:35.277",
         "flowFileRepositoryStorageUsage": {
            "freeSpace": "286.93 GB",
            "totalSpace": "464.78 GB",
            "usedSpace": "177.85 GB",
            "freeSpaceBytes": 308090789888,
            "totalSpaceBytes": 499057160192,
            "usedSpaceBytes": 190966370304,
            "utilization": "38.0%"
         },
         "contentRepositoryStorageUsage": [
            {
               "identifier": "default",
               "freeSpace": "286.93 GB",
               "totalSpace": "464.78 GB",
               "usedSpace": "177.85 GB",
               "freeSpaceBytes": 308090789888,
               "totalSpaceBytes": 499057160192,
               "usedSpaceBytes": 190966370304,
               "utilization": "38.0%"
            }
         ],
         "provenanceRepositoryStorageUsage": [
            {
               "identifier": "default",
               "freeSpace": "286.93 GB",
               "totalSpace": "464.78 GB",
               "usedSpace": "177.85 GB",
               "freeSpaceBytes": 308090789888,
               "totalSpaceBytes": 499057160192,
               "usedSpaceBytes": 190966370304,
               "utilization": "38.0%"
            }
         ],
         "garbageCollection": [
            {
               "name": "G1 Young Generation",
               "collectionCount": 344,
               "collectionTime": "00:00:06.239",
               "collectionMillis": 6239
            },
            {
               "name": "G1 Old Generation",
               "collectionCount": 0,
               "collectionTime": "00:00:00.000",
               "collectionMillis": 0
            }
         ],
         "statsLastRefreshed": "09:30:20 SGT",
         "versionInfo": {
            "niFiVersion": "1.7.1",
            "javaVendor": "Oracle Corporation",
            "javaVersion": "1.8.0_151",
            "osName": "Windows 7",
            "osVersion": "6.1",
            "osArchitecture": "amd64",
            "buildTag": "nifi-1.7.1-RC1",
            "buildTimestamp": "07/12/2018 12:54:43 SGT"
         }
      }
   }
}

Apache NiFi - 升级

在开始升级 Apache NiFi 之前，请阅读发行说明以了解更改和新增功能。用户需要评估这些新增功能和更改对其当前 NiFi 安装的影响。以下是从 Apache NiFi 新版本获取发行说明的链接：

https://cwiki.apache.org/confluence/display/NIFI/Release+Notes

在集群设置中，用户需要升级集群中每个节点的 NiFi 安装。请按照以下步骤升级 Apache NiFi。

• 备份当前 NiFi 中存在的 lib 或任何其他文件夹中的所有自定义 NAR。

• 下载新版本的 Apache NiFi。以下是从最新 NiFi 版本下载源代码和二进制文件的链接。

  https://nifi.apache.org/download.html

• 在当前 NiFi 的同一安装目录中创建一个新目录，并将新版本的 Apache NiFi 解压缩到该目录。

• 优雅地停止 NiFi。首先停止所有处理器，并让数据流中存在的 FlowFile 都被处理完成。一旦没有更多 FlowFile，停止 NiFi。

• 将当前 NiFi 安装的 authorizers.xml 配置复制到新版本。

• 从当前版本更新新 NiFi 版本的 bootstrap-notification-services.xml 和 bootstrap.conf 中的值。

• 将 logback.xml 中的自定义日志记录添加到新 NiFi 安装中。

• 从当前版本配置 login-identity-providers.xml 中的登录身份提供程序。

• 从当前版本更新新 NiFi 安装的 nifi.properties 中的所有属性。

• 请确保新版本的组和用户与当前版本相同，以避免任何权限被拒绝错误。

• 将当前版本的 state-management.xml 配置复制到新版本。

• 将以下目录的内容从当前版本的 NiFi 安装复制到新版本中的相同目录。

  • ./conf/flow.xml.gz

  • 以及 archive 目录中的 flow.xml.gz。

  • 对于溯源和内容存储库，请在 nifi.properties 文件中将值更改为当前存储库。

  • 从 ./state/local 复制状态，或如果指定了任何其他外部目录，则在 nifi.properties 中进行更改。

• 重新检查所有执行的更改，并检查它们是否对新 NiFi 版本中添加的任何新更改有影响。如果有任何影响，请检查解决方案。

• 启动所有 NiFi 节点，并验证所有数据流是否正常工作，存储库是否正在存储数据，并且 UI 是否在没有错误的情况下检索数据。

• 监控公告板一段时间，以检查是否有任何新的错误。

• 如果新版本正常工作，则可以将当前版本存档并从目录中删除。

Apache NiFi - 远程处理组

Apache NiFi 远程进程组 (RPG) 使数据流能够使用站点到站点协议将数据流中的 FlowFile 指向不同的 NiFi 实例。截至 1.7.1 版本，NiFi 不提供负载均衡关系，因此 RPG 用于在 NiFi 数据流中进行负载均衡。

开发人员可以通过将 NiFi UI 顶部工具栏中的图标拖动到画布上（如上图所示）来添加 RPG。要配置 RPG，开发人员必须添加以下字段：

开发人员需要在使用之前启用它，就像我们使用处理器之前启动它们一样。

Apache NiFi - 控制器设置

Apache NiFi 提供共享服务，处理器和报告任务可以使用这些服务，称为控制器设置。这些就像数据库连接池，可以被访问相同数据库的处理器使用。

要访问控制器设置，请使用 NiFi UI 右上角的下拉菜单，如下面的图像所示。

Apache NiFi 提供了许多控制器设置，我们将讨论一个常用的设置以及如何在 NiFi 中进行设置。

DBCPConnectionPool

单击“控制器设置”选项后，在 NiFi 设置页面中添加加号。然后从控制器设置列表中选择 DBCPConnectionPool。DBCPConnectionPool 将添加到主 NiFi 设置页面，如下面的图像所示。

它包含以下有关控制器 **设置：名称** 的信息：

• 类型
• 捆绑包
• 状态
• 范围
• 配置和删除图标

单击“配置”图标并填写所需字段。这些字段列在下表中：

要停止或配置控制器设置，首先必须停止所有附加的 NiFi 组件。NiFi 还为控制器设置添加了范围以管理其配置。因此，只有共享相同设置的组件不会受到影响，并将使用相同的控制器设置。

Apache NiFi - 报告任务

Apache NiFi 报告任务类似于在后台运行并发送或记录 NiFi 实例统计信息的控制器服务。NiFi 报告任务也可以从与控制器设置相同的页面访问，但在不同的选项卡中。

要添加报告任务，开发人员需要点击报告任务页面右上角的加号按钮。这些报告任务主要用于监控 NiFi 实例的活动，无论是公告还是来源。主要这些报告任务使用站点到站点将 NiFi 统计数据传输到其他节点或外部系统。

现在让我们添加一个已配置的报告任务以了解更多信息。

MonitorMemory

此报告任务用于在内存池超过指定百分比时生成公告。请按照以下步骤配置 MonitorMemory 报告任务 -

• 添加加号并在列表中搜索 MonitorMemory。

• 选择 MonitorMemory 并点击添加。

• 一旦它添加到报告任务主页面，点击配置图标。

• 在属性选项卡中，选择要监控的内存池。

• 选择希望公告提醒用户的百分比。

• 启动报告任务。

Apache NiFi - 自定义处理器

Apache NiFi 是一个开源平台，允许开发人员在 NiFi 库中添加自定义处理器。请按照以下步骤创建自定义处理器。

• 从下面给出的链接下载 Maven 最新版本。

  https://maven.apache.org/download.cgi

• 添加名为 M2_HOME 的环境变量，并将值设置为 maven 的安装目录。

• 从以下链接下载 Eclipse IDE。

  https://www.eclipse.org/downloads/download.php

• 打开命令提示符并执行 Maven Archetype 命令。

> mvn archetype:generate

• 在原型项目中搜索 nifi 类型。

• 选择 org.apache.nifi:nifi-processor-bundle-archetype 项目。

• 然后从版本列表中选择最新版本，即本教程的 1.7.1。

• 输入 groupId、artifactId、version、package 和 artifactBaseName 等。

• 然后将创建一个包含两个目录的 Maven 项目。

  • nifi-<artifactBaseName>-processors

  • nifi-<artifactBaseName>-nar

• 在 nifi-<artifactBaseName>-processors 目录中运行以下命令以将项目添加到 eclipse 中。

mvn install eclipse:eclipse

• 打开 eclipse 并从文件菜单中选择导入。

• 然后选择“将现有项目导入工作区”，并将项目从 nifi-<artifactBaseName>-processors 目录添加到 eclipse 中。

• 在 public void onTrigger(ProcessContext context, ProcessSession session) 函数中添加您的代码，该函数在每次计划处理器运行时运行。

• 然后通过运行下面提到的命令将代码打包到 NAR 文件中。

mvn clean install

• NAR 文件将在 nifi- 中创建-nar/target 目录。

• 将 NAR 文件复制到 Apache NiFi 的 lib 文件夹并重新启动 NiFi。

• NiFi 成功重启后，检查处理器列表中是否有新的自定义处理器。

• 对于任何错误，请检查 ./logs/nifi.log 文件。

Apache NiFi - 自定义控制器服务

Apache NiFi 是一个开源平台，允许开发人员在 Apache NiFi 中添加自定义控制器服务。步骤和工具与创建自定义处理器几乎相同。

• 打开命令提示符并执行 Maven Archetype 命令。

> mvn archetype:generate

• 在原型项目中搜索 nifi 类型。

• 选择org.apache.nifi:nifi-service-bundle-archetype项目。

• 然后从版本列表中选择最新版本 - 本教程的 1.7.1。

• 输入 groupId、artifactId、version、package 和 artifactBaseName 等。

• 将创建一个包含目录的 Maven 项目。

  • nifi-<artifactBaseName>

  • nifi-<artifactBaseName>-nar

  • nifi-<artifactBaseName>-api

  • nifi-<artifactBaseName>-api-nar

• 在 nifi-<artifactBaseName> 和 nifi-<artifactBaseName>-api 目录中运行以下命令以将这两个项目添加到 eclipse 中。

  • mvn install eclipse:eclipse

• 打开 eclipse 并从文件菜单中选择导入。

• 然后选择“将现有项目导入工作区”，并将项目从 nifi-<artifactBaseName> 和 nifi-<artifactBaseName>-api 目录添加到 eclipse 中。

• 在源文件中添加您的代码。

• 然后通过运行下面提到的命令将代码打包到 NAR 文件中。

  • mvn clean install

• 将在每个 nifi-<artifactBaseName>/target 和 nifi-<artifactBaseName>-api/target 目录中创建两个 NAR 文件。

• 将这些 NAR 文件复制到 Apache NiFi 的 lib 文件夹并重新启动 NiFi。

• NiFi 成功重启后，检查处理器列表中是否有新的自定义处理器。

• 对于任何错误，请检查./logs/nifi.log文件。

Apache NiFi - 日志

Apache NiFi 使用 logback 库来处理其日志记录。NiFi 的 conf 目录中有一个名为 logback.xml 的文件，用于配置 NiFi 中的日志记录。日志生成在 NiFi 的 logs 文件夹中，日志文件如下所述。

nifi-app.log

这是 nifi 的主日志文件，记录了从 NAR 文件加载到 NiFi 组件遇到的运行时错误或公告等所有 apache NiFi 应用程序活动。以下是logback.xml文件中nifi-app.log文件的默认附加程序。

<appender name="APP_FILE"
class="ch.qos.logback.core.rolling.RollingFileAppender">
   <file>${org.apache.nifi.bootstrap.config.log.dir}/nifi-app.log</file>
   <rollingPolicy
      class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
      <fileNamePattern>
         ${org.apache.nifi.bootstrap.config.log.dir}/
	      nifi-app_%d{yyyy-MM-dd_HH}.%i.log
      </fileNamePattern>
      <maxFileSize>100MB</maxFileSize>
      <maxHistory>30</maxHistory>
   </rollingPolicy>
   <immediateFlush>true</immediateFlush>
   <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
   </encoder>
</appender>

附加程序名称为 APP_FILE，类为 RollingFileAppender，这意味着记录器使用回滚策略。默认情况下，最大文件大小为 100 MB，可以更改为所需大小。APP_FILE 的最大保留时间为 30 个日志文件，可以根据用户需求进行更改。

nifi-user.log

此日志包含用户事件，如 Web 安全、Web API 配置、用户授权等。以下是 logback.xml 文件中 nifi-user.log 的附加程序。

<appender name="USER_FILE"
   class="ch.qos.logback.core.rolling.RollingFileAppender">
   <file>${org.apache.nifi.bootstrap.config.log.dir}/nifi-user.log</file>
   <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>
         ${org.apache.nifi.bootstrap.config.log.dir}/
	      nifi-user_%d.log
      </fileNamePattern>
      <maxHistory>30</maxHistory>
   </rollingPolicy>
   <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
   </encoder>
</appender>

附加程序名称为 USER_FILE。它遵循回滚策略。USER_FILE 的最大保留期为 30 个日志文件。以下是 nifi-user.log 中存在的 USER_FILE 附加程序的默认记录器。

<logger name="org.apache.nifi.web.security" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.web.api.config" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.authorization" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.cluster.authorization" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.web.filter.RequestLogger" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>

nifi-bootstrap.log

此日志包含引导日志、apache NiFi 的标准输出（代码中编写的全部 system.out，主要用于调试）和标准错误（代码中编写的全部 system.err）。以下是 logback.log 中 nifi-bootstrap.log 的默认附加程序。

<appender name="BOOTSTRAP_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
   <file>${org.apache.nifi.bootstrap.config.log.dir}/nifi-bootstrap.log</file>
   <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>
         ${org.apache.nifi.bootstrap.config.log.dir}/nifi-bootstrap_%d.log
      </fileNamePattern>
      <maxHistory>5</maxHistory>
   </rollingPolicy>
   <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
   </encoder>
</appender>

nifi-bootstrap.log 文件的附加程序名称为 BOOTSTRAP_FILE，它也遵循回滚策略。BOOTSTRAP_FILE 附加程序的最大保留时间为 5 个日志文件。以下是 nifi-bootstrap.log 文件的默认记录器。

<logger name="org.apache.nifi.bootstrap" level="INFO" additivity="false">
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>
<logger name="org.apache.nifi.bootstrap.Command" level="INFO" additivity="false">
   <appender-ref ref="CONSOLE" />
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>
<logger name="org.apache.nifi.StdOut" level="INFO" additivity="false">
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>
<logger name="org.apache.nifi.StdErr" level="ERROR" additivity="false">
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>