Apache NiFi 入门指南

Posted 2020-11-27 victor的博客

 

本指南使用于谁？

本指南适用于从未使用过，在NiFi中有限度接触或仅完成特定任务的用户。本指南不是详尽的说明手册或参考指南。“ 用户指南”提供了大量信息，旨在提供更加详尽的资源，并且作为参考指南非常有用。相比之下，本指南旨在为用户提供所需的信息，以便了解如何使用NiFi，以便快速轻松地构建强大而灵活的数据流。

一些因为本指南中的某些信息仅适用于初次使用的用户，而其他信息可能适用于那些使用过NiFi的人，本指南分为几个不同的部分，其中一些可能对某些部分没用读者。随意跳转到最适合您的部分。

本指南确实希望用户基本了解NiFi是什么，并且没有深入研究这个细节。可以在Overview文档中找到此级别的信息 。

本指南中使用的术语

为了谈论NiFi，读者应该熟悉一些关键术语。我们将在此高度解释这些特定于NiFi的术语。

FlowFile：每条“用户数据”（即用户带入NiFi进行处理和分发的数据）称为FlowFile。FlowFile由两部分组成：属性和内容。内容是用户数据本身。属性是与用户数据关联的 key-value 对。 

Processor：处理器是NiFi组件，负责创建，发送，接收，转换，路由，拆分，合并和处理FlowFiles。它是NiFi用户可用于构建其数据流的最重要的组件。

下载和安装NiFi

NiFi可以从NiFi下载页面下载。有两种可用的打包选项：针对Linux定制的“tarball”和更适用于Windows用户的zip文件。Mac OS X用户也可以使用tarball或通过Homebrew安装。

要通过Homebrew安装，只需运行命令即可brew install nifi。

对于未运行OS X或未安装Homebrew的用户，在下载您想要使用的NiFi版本后，只需将存档解压缩到您希望运行该应用程序的位置。

有关如何配置NiFi实例的信息（例如，配置安全性，数据存储配置或运行NiFi的端口），请参阅“ 管理员指南”。

启动NiFi

一旦按照上述方式下载并安装了NiFi，就可以使用适合您操作系统的机制启动它。

对于Windows用户

对于Windows用户，请导航到安装NiFi的文件夹。在此文件夹中有一个名为bin的子文件夹。导航到此子文件夹，然后双击该run-nifi.bat文件。

这将启动NiFi并让它在前台运行。要关闭NiFi，请选择已启动的窗口，并在按下C的同时按住Ctrl键。

适用于Linux / Mac OS X用户

对于Linux和OS X用户，使用终端窗口导航到安装NiFi的目录。要在前台运行NiFi，请运行bin/nifi.sh run。这将使应用程序继续运行，直到用户按下Ctrl-C。那时，它将启动应用程序的关闭。

要在后台运行NiFi，请运行bin/nifi.sh start。这将启动应用程序以开始运行。要检查状态并查看NiFi当前是否正在运行，请执行该命令bin/nifi.sh status。可以通过执行命令关闭NiFi bin/nifi.sh stop。

作为服务安装

目前，仅支持Linux和Mac OS X用户安装NiFi作为服务。要将应用程序作为服务安装，请导航到“终端”窗口中的安装目录，然后执行命令bin/nifi.sh install 以使用默认名称安装服务nifi。要为服务指定自定义名称，请使用可选的第二个参数（该服务的名称）执行该命令。例如，要将NiFi作为具有名称的服务安装dataflow，请使用该命令bin/nifi.sh install dataflow。

安装后，可以使用适当的命令启动和停止服务，例如sudo service nifi start 和sudo service nifi stop。此外，可以通过检查运行状态sudo service nifi status。

我开始使用NiFi。怎么办？

现在NiFi已经启动，我们可以调出用户界面（UI）来创建和监控我们的数据流。要开始使用，请打开Web浏览器并导航到http://localhost:8080/nifi。可以通过编辑nifi.propertiesNiFi conf目录中的文件来更改端口，但默认端口为8080。

这将打开用户界面，此时这是一个用于编排数据流的空白画布：

 

 

用户界面有多种工具可用于创建和管理您的第一个数据流：

全局菜单包含以下选项：

添加处理器

我们现在可以通过在画布中添加Processor来开始创建数据流。要执行此操作，请将处理器图标（）从屏幕左上方拖动到画布中间（图纸类背景）并将其放在那里。这将为我们提供一个对话框，允许我们选择要添加的处理器：

我们有很多选择可供选择。为了成为系统的导向，假设我们只想将本地磁盘中的文件导入NiFi。当开发人员创建处理器时，开发人员可以为该处理器分配“标签”。这些可以被认为是关键字。我们可以通过在对话框右上角的“过滤器”框中键入来按这些标记或处理器名称进行过滤。想要从本地磁盘中提取文件时，请键入您想到的关键字。例如，键入关键字“file”将为我们提供一些处理文件的不同处理器。按“本地”一词过滤也会很快缩小列表范围。如果我们从列表中选择处理器，我们将在对话框底部附近看到处理器的简要说明。这应该告诉我们处理器的确切功能。GetFile Processor 的描述 告诉我们它将数据从我们的本地磁盘拉入NiFi，然后删除本地文件。然后我们可以双击处理器类型或选择它并选择 Add按钮。处理器将被添加到画布中删除它的位置。

 

配置处理器

现在我们已经添加了GetFile处理器，我们可以通过右键单击Processor并选择Configure菜单项来配置它。提供的对话框允许我们配置许多可在“ 用户指南”中阅读的不同选项，但为了本指南，我们将重点关注“属性”选项卡。选择Properties选项卡后，我们将获得一个可以为Processor配置的几个不同属性的列表。可用的属性取决于处理器的类型，并且每种类型通常都不同。粗体属性是必需属性。在配置完所有必需属性之前，无法启动处理器。为GetFile配置的最重要的属性是从中获取文件的目录。如果我们将目录名设置为./data-in，这将导致处理器开始拾取中的任何数据 data-in NiFi主目录的子目录。我们可以选择为此处理器配置多个不同的属性。如果不确定特定属性的作用，我们可以将鼠标悬停在 属性名称旁边的帮助图标（ ）上，以便阅读该属性的描述。此外，将鼠标悬停在“帮助”图标上时显示的工具提示将提供该属性的默认值（如果存在），有关该属性是否支持表达式语言的信息（请参阅下面的“ 属性值”部分中的“ 表达式语言/使用属性”）），以及之前为该属性配置的值。

data-in要使此属性有效，请创建在NiFi主目录中命名的目录，然后单击Ok按钮以关闭对话框。

连接处理器

每个处理器都有一组定义的“Relationships”，它能够将数据发送到。处理器完成FlowFile的处理后，会将其传输到其中一个关系。这允许用户根据Processing的结果配置如何处理FlowFiles。例如，许多处理器定义了两个关系：success和failure。然后，如果处理器能够成功处理数据并且如果处理器由于某种原因无法处理数据而以完全不同的方式将数据路由通过流，则用户能够配置要通过流单向路由的数据。或者，根据用例，它可以简单地将两个关系路由到流中的相同路由。

现在我们已经添加并配置了我们的GetFile处理器并应用了配置，我们可以在处理器的左上角看到一个警告图标（  ），表示处理器未处于有效状态。将鼠标悬停在此图标上，我们可以看到该success 关系尚未定义。这只是意味着我们没有告诉NiFi如何处理处理器转移到success关系的数据。

为了解决这个问题，让我们按照上面的相同步骤添加另一个可以连接GetFile处理器的处理器。但是，这一次，我们只需记录FlowFile存在的属性。为此，我们将添加一个LogAttributes处理器。

我们现在可以将GetFile处理器的输出发送到LogAttribute Processor。用鼠标悬停在GetFile处理器上，连接图标（  ）将出现在处理器的中间。我们可以将此图标从GetFile处理器拖到LogAttribute处理器。这为我们提供了一个对话框，用于选择我们要为此连接包含哪些关系。由于GetFile只有一个关系success，因此会自动为我们选择。

单击“设置”选项卡提供了一些用于配置此连接的行为方式的选项：

如果我们愿意，我们可以给Connection一个名字。否则，连接名称将基于所选的关系。我们还可以设置数据的到期时间。默认情况下，它设置为“0秒”，表示数据不应过期。但是，我们可以更改该值，以便当此Connection中的数据达到特定年龄时，它将自动删除（并且将创建相应的EXPIRE Provenance事件）。

背压阈值允许我们指定在不再安排源处理器运行之前允许队列的完整程度。这使我们能够处理一个处理器能够比下一个处理器能够消耗该数据更快地生成数据的情况。如果在整个过程中为每个连接配置了背压，则将数据引入系统的处理器最终将经历背压并停止引入新数据，以便我们的系统能够恢复。

最后，我们在右侧有优先顺序。这允许我们控制如何排序此队列中的数据。我们可以将优先级从“可用的优先级排序器”列表拖到“选定的优先级排序器”列表中，以激活优先级排序器。如果激活了多个优先级排序器，则将对它们进行评估，以便首先评估首先列出的优先级排序器，如果根据该优先级排序器确定两个FlowFiles相等，则将使用第二个优先级排序器。

为了便于讨论，我们只需单击Add即可将Connection添加到图表中。我们现在应该看到Alert图标已更改为Stopped图标（  ）。但是，LogAttribute Processor现在无效，因为它的success关系尚未连接到任何东西。让我们通过发信号通知success由LogAttribute 路由到的数据应该是“自动终止” 来解决这个问题，这意味着NiFi应该考虑FlowFile的处理完成并“丢弃”数据。为此，我们配置LogAttribute Processor。在“设置”选项卡的右侧，我们可以选中“ success自动终止关系数据”旁边的框。单击OK将关闭对话框并显示两个处理器现已停止。

启动和停止处理器

此时，我们的图表上有两个处理器，但没有发生任何事情。为了启动处理器，我们可以单独单击每个处理器，然后右键单击并选择Start菜单项。或者，我们可以选择第一个处理器，然后在选择其他处理器的同时按住Shift键以选择两者。然后，我们可以右键单击并选择Start菜单项。作为使用上下文菜单的替代方法，我们可以选择处理器，然后单击“操作”调板中的“开始”图标。

启动后，处理器左上角的图标将从停止的图标更改为正在运行的图标。然后，我们可以使用“操作”调板或Stop菜单项中的“停止”图标来停止处理器。

处理器启动后，我们无法再配置它。相反，当我们右键单击处理器时，我们可以选择查看其当前配置。为了配置处理器，我们必须首先停止处理器并等待可能正在执行的任何任务完成。当前正在执行的任务数显示在处理器的右上角附近，但如果当前没有任务，则不会显示任何内容。

获取处理器的更多信息

由于每个处理器都能够公开多个不同的属性和关系，因此记住每个处理器的所有不同部分的工作原理可能很困难。要解决此问题，您可以右键单击处理器并选择Usage菜单项。这将为您提供处理器的使用信息，例如处理器的描述，可用的不同关系，何时使用不同的关系，处理器及其文档公开的属性，以及哪些FlowFile属性（在传入的FlowFiles上有预期的，以及哪些属性（如果有的话）被添加到传出的FlowFiles中。

其他组件

用户可以将处理器拖放到图表上的工具栏包括可用于构建数据流的其他几个组件。这些组件包括输入和输出端口，漏斗，进程组和远程进程组。由于本文档的预期范围，我们不会在此讨论这些元素，但可以在“ 用户指南”的“ 构建数据流”部分中找到相关信息 。

什么处理器可用

为了创建有效的数据流，用户必须了解可用的处理器类型。NiFi包含许多不同的处理器。这些处理器提供了从众多不同系统中提取数据，路由，转换，处理，拆分和聚合数据以及将数据分发到多个系统的功能。

几乎每个NiFi版本中可用的处理器数量都在增加。因此，我们不会尝试命名每个可用的处理器，但我们将重点介绍一些最常用的处理器，按功能对它们进行分类。

数据转换

• CompressContent：压缩或解压缩内容

• ConvertCharacterSet：将用于编码内容的字符集从一个字符集转换为另一个字符集

• EncryptContent：加密或解密内容

• ReplaceText：使用正则表达式修改文本内容

• TransformXml：将XSLT转换应用于XML内容

• JoltTransformJSON：应用JOLT规范来转换JSON内容

路由和调解

• ControlRate：限制数据流经流量的一部分的速率

• DetectDuplicate：根据一些用户定义的标准监视重复的FlowFiles。通常与HashContent一起使用

• DistributeLoad：通过仅将一部分数据分发到每个用户定义的关系来加载平衡或样本数据

• MonitorActivity：当用户定义的时间段过去时发送通知，而没有任何数据通过流中的特定点。（可选）在数据流恢复时发送通知。

• RouteOnAttribute：根据FlowFile包含的属性路由FlowFile。

• ScanAttribute：扫描FlowFile上用户定义的属性集，检查是否有任何属性与用户定义的字典中找到的术语匹配。

• RouteOnContent：搜索FlowFile的内容，以查看它是否与任何用户定义的正则表达式匹配。如果是，则FlowFile将路由到已配置的关系。

• ScanContent：搜索FlowFile的内容，以查找用户定义的字典中存在的术语，并根据这些术语的存在与否进行路由。字典可以包含文本条目或二进制条目。

• ValidateXml：针对XML模式验证XML内容; 根据用户定义的XML Schema，FlowFile的内容是否有效，路由FlowFile。

数据库访问

• ConvertJSONToSQL：将JSON文档转换为SQL INSERT或UPDATE命令，然后可以将其传递给PutSQL Processor

• ExecuteSQL：执行用户定义的SQL SELECT命令，将结果写入Avro格式的FlowFile

• PutSQL：通过执行FlowFile内容定义的SQL DDM语句来更新数据库

• SelectHiveQL：对Apache Hive数据库执行用户定义的HiveQL SELECT命令，将结果写入Avro或CSV格式的FlowFile

• PutHiveQL：通过执行FlowFile内容定义的HiveQL DDM语句来更新Hive数据库

属性提取

• EvaluateJsonPath：用户提供JSONPath表达式（类似于XPath，用于XML解析/提取），然后根据JSON内容评估这些表达式，以替换FlowFile内容或将值提取到用户命名的Attribute中。

• EvaluateXPath：用户提供XPath表达式，然后根据XML内容评估这些表达式，以替换FlowFile内容或将值提取到用户命名的Attribute中。

• EvaluateXQuery：用户提供XQuery查询，然后根据XML内容评估此查询，以替换FlowFile内容或将值提取到用户命名的Attribute中。

• ExtractText：用户提供一个或多个正则表达式，然后根据FlowFile的文本内容对其进行评估，然后将提取的值添加为用户命名的属性。

• HashAttribute：对用户定义的现有属性列表的串联执行散列函数。

• HashContent：对FlowFile的内容执行散列函数，并将散列值添加为Attribute。

• IdentifyMimeType：评估FlowFile的内容，以确定FlowFile封装的文件类型。此处理器能够检测许多不同的MIME类型，例如图像，文字处理器文档，文本和压缩格式，仅举几例。

• UpdateAttribute：向FlowFile添加或更新任意数量的用户定义属性。这对于添加静态配置的值以及使用表达式语言动态地派生属性值非常有用。该处理器还提供“高级用户界面”，允许用户根据用户提供的规则有条件地更新属性。

系统交互

• ExecuteProcess：运行用户定义的操作系统命令。进程的StdOut被重定向，以便写入StdOut的内容成为出站FlowFile的内容。此处理器是源处理器 - 其输出预计会生成新的FlowFile，并且系统调用预计不会接收任何输入。要为进程提供输入，请使用ExecuteStreamCommand Processor。

• ExecuteStreamCommand：运行用户定义的操作系统命令。FlowFile的内容可选地流式传输到进程的StdIn。写入StdOut的内容将成为hte出站FlowFile的内容。此处理器不能用作源处理器 - 必须输入传入的FlowFiles才能执行其工作。要使用源处理器执行相同类型的功能，请参阅ExecuteProcess Processor。

数据摄取

• GetFile：将文件内容从本地磁盘（或网络连接的磁盘）流式传输到NiFi，然后删除原始文件。此处理器应将文件从一个位置移动到另一个位置，而不是用于复制数据。

• GetFTP：通过FTP将远程文件的内容下载到NiFi中，然后删除原始文件。期望此处理器将数据从一个位置移动到另一个位置，而不是用于复制数据。

• GetSFTP：通过SFTP将远程文件的内容下载到NiFi中，然后删除原始文件。期望此处理器将数据从一个位置移动到另一个位置，而不是用于复制数据。

• GetJMSQueue：从JMS队列下载消息，并根据JMS消息的内容创建FlowFile。可选地，JMS属性也可以作为属性复制。

• GetJMSTopic：从JMS主题下载消息，并根据JMS消息的内容创建FlowFile。可选地，JMS属性也可以作为属性复制。此处理器支持持久订阅和非持久订阅。

• GetHTTP：将基于HTTP或HTTPS的远程URL的内容下载到NiFi中。处理器将记住ETag和Last-Modified Date，以确保不会持续摄取数据。

• ListenHTTP：启动HTTP（或HTTPS）服务器并侦听传入连接。对于任何传入的POST请求，请求的内容将作为FlowFile写出，并返回200响应。

• ListenUDP：侦听传入的UDP数据包，并为每个数据包或每个数据包创建一个FlowFile（取决于配置），并将FlowFile发送到“成功”关系。

• GetHDFS：监视HDFS中用户指定的目录。每当新文件进入HDFS时，它将被复制到NiFi并从HDFS中删除。此处理器应将文件从一个位置移动到另一个位置，而不是用于复制数据。如果在群集中运行，预计此处理器也仅在主节点上运行。要从HDFS复制数据并使其保持原状，或者从群集中的多个节点流式传输数据，请参阅ListHDFS处理器。

• ListHDFS / FetchHDFS：ListHDFS监视HDFS中用户指定的目录，并发出一个FlowFile，其中包含它遇到的每个文件的文件名。然后，它通过分布式缓存在整个NiFi集群中保持此状态。然后可以在群集中扇出这些FlowFile并将其发送到FetchHDFS处理器，后者负责获取这些文件的实际内容并发出包含从HDFS获取的内容的FlowFiles。

• FetchS3Object：从Amazon Web Services（AWS）简单存储服务（S3）获取对象的内容。出站FlowFile包含从S3接收的内容。

• GetKafka：从Apache Kafka获取消息，特别是0.8.x版本。消息可以作为每个消息的FlowFile发出，也可以使用用户指定的分隔符一起批处理。

• GetMongo：对MongoDB执行用户指定的查询，并将内容写入新的FlowFile。

• GetTwitter：允许用户注册过滤器以收听Twitter“花园软管”或企业端点，为收到的每条推文创建一个FlowFile。

数据出口/发送数据

• PutEmail：向配置的收件人发送电子邮件。FlowFile的内容可选择作为附件发送。

• PutFile：将FlowFile的内容写入本地（或网络连接）文件系统上的目录。

• PutFTP：将FlowFile的内容复制到远程FTP服务器。

• PutSFTP：将FlowFile的内容复制到远程SFTP服务器。

• PutJMS：将FlowFile的内容作为JMS消息发送到JMS代理，可选择根据Attributes添加JMS属性。

• PutSQL：将FlowFile的内容作为SQL DDL语句（INSERT，UPDATE或DELETE）执行。FlowFile的内容必须是有效的SQL语句。属性可以用作参数，以便FlowFile的内容可以是参数化的SQL语句，以避免SQL注入攻击。

• PutKafka：将FlowFile的内容作为消息发送到Apache Kafka，特别是0.8.x版本。FlowFile可以作为单个消息或分隔符发送，例如可以指定换行符，以便为单个FlowFile发送许多消息。

• PutMongo：将FlowFile的内容作为INSERT或UPDATE发送到Mongo。

分裂和聚合

• SplitText：SplitText接收单个FlowFile，其内容为文本，并根据配置的行数将其拆分为1个或多个FlowFiles。例如，可以将处理器配置为将FlowFile拆分为多个FlowFile，每个FlowFile只有一行。

• SplitJson：允许用户将包含数组或许多子对象的JSON对象拆分为每个JSON元素的FlowFile。

• SplitXml：允许用户将XML消息拆分为多个FlowFiles，每个FlowFiles包含原始段。这通常在多个XML元素与“wrapper”元素连接在一起时使用。然后，此处理器允许将这些元素拆分为单独的XML元素。

• UnpackContent：解压缩不同类型的存档格式，例如ZIP和TAR。然后，归档中的每个文件都作为单个FlowFile传输。

• MergeContent：此处理器负责将许多FlowFiles合并到一个FlowFile中。可以通过将其内容与可选的页眉，页脚和分界符连接在一起，或者通过指定存档格式（如ZIP或TAR）来合并FlowFiles。FlowFiles可以根据公共属性进行分箱，或者如果它们被其他拆分过程拆分，则可以进行“碎片整理”。根据元素的数量或FlowFiles内容的总大小，每个bin的最小和最大大小是用户指定的，并且还可以分配可选的Timeout，以便FlowFiles只等待其bin变为满多少时间。

• SegmentContent：根据某些已配置的数据大小将FlowFile划分为可能的许多较小的FlowFile。不对任何类型的分界符执行拆分，而是仅基于字节偏移执行拆分。这是在传输FlowFiles之前使用的，以便通过并行发送许多不同的部分来提供更低的延迟。另一方面，MergeContent处理器可以使用碎片整理模式重新组装这些FlowFiles。

• SplitContent：将单个FlowFile拆分为可能的许多FlowFile，类似于SegmentContent。但是，对于SplitContent，不会在任意字节边界上执行拆分，而是指定要拆分内容的字节序列。

HTTP

• GetHTTP：将基于HTTP或HTTPS的远程URL的内容下载到NiFi中。处理器将记住ETag和Last-Modified Date，以确保不会持续摄取数据。

• ListenHTTP：启动HTTP（或HTTPS）服务器并侦听传入连接。对于任何传入的POST请求，请求的内容将作为FlowFile写出，并返回200响应。

• InvokeHTTP：执行用户配置的HTTP请求。此处理器比GetHTTP和PostHTTP更通用，但需要更多配置。此处理器不能用作源处理器，并且需要具有传入的FlowFiles才能被触发以执行其任务。

• PostHTTP：执行HTTP POST请求，将FlowFile的内容作为消息正文发送。这通常与ListenHTTP结合使用，以便在无法使用站点到站点的情况下在两个不同的NiFi实例之间传输数据（例如，当节点无法直接访问并且能够通过HTTP进行通信时代理）。 注意：除了现有的RAW套接字传输之外，HTTP还可用作站点到站点传输协议。它还支持HTTP代理。建议使用HTTP站点到站点，因为它更具可扩展性，并且可以使用具有更好用户身份验证和授权的输入/输出端口提供双向数据传输。

• HandleHttpRequest / HandleHttpResponse：HandleHttpRequest Processor是一个源处理器，与ListenHTTP类似，启动嵌入式HTTP（S）服务器。但是，它不会向客户端发送响应。相反，FlowFile与HTTP请求的主体一起作为其内容和属性发送，作为属性的所有典型Servlet参数，标题等。然后，HandleHttpResponse能够在FlowFile完成处理后将响应发送回客户端。这些处理器总是希望彼此结合使用，并允许用户在NiFi中可视化地创建Web服务。这对于将前端添加到非基于Web的协议或围绕已经由NiFi执行的某些功能（例如数据格式转换）添加简单的Web服务特别有用。

亚马逊网络服务

• FetchS3Object：获取存储在Amazon Simple Storage Service中的对象的内容（S3）。然后，从S3检索的内容将写入FlowFile的内容。

• PutS3Object：使用配置的凭据，密钥和存储桶名称将FlowFile的内容写入Amazon S3对象。

• PutSNS：将FlowFile的内容作为通知发送到Amazon Simple Notification Service（SNS）。

• GetSQS：从Amazon Simple Queuing Service（SQS）中提取消息，并将消息内容写入FlowFile的内容。

• PutSQS：将FlowFile的内容作为消息发送到Amazon Simple Queuing Service（SQS）。

• DeleteSQS：从Amazon Simple Queuing Service（SQS）中删除消息。这可以与GetSQS一起使用，以便从SQS接收消息，对其执行一些处理，然后仅在成功完成处理后才从队列中删除该对象。

使用属性

每个FlowFile都使用多个属性创建，这些属性将在FlowFile的生命周期中发生变化。FlowFile的概念非常强大，并提供三个主要优点。首先，它允许用户在流中做出路由决策，以便满足某些条件的FlowFiles可以与其他FlowFiles不同地进行处理。这是使用RouteOnAttribute和类似的处理器完成的。

其次，使用属性以便以这样的方式配置处理器：处理器的配置依赖于数据本身。例如，PutFile Processor能够使用Attributes来知道每个FlowFile的存储位置，而每个FlowFile的目录和文件名属性可能不同。

最后，属性提供了有关数据的极有价值的上下文。在查看FlowFile的Provenance数据时，这非常有用。这允许用户搜索符合特定条件的Provenance数据，并且还允许用户在检查Provenance事件的详细信息时查看此上下文。通过这样做，用户然后能够获得关于为什么以这样或那样的方式处理数据的有价值的见解，简单地通过浏览与内容一起携带的该上下文。

共同属性

每个FlowFile都有一组最小的属性：

• filename：可用于将数据存储到本地或远程文件系统的文件名。

• path：可用于将数据存储到本地或远程文件系统的目录的名称。

• uuid：一个通用唯一标识符，用于区分FlowFile与系统中的其他FlowFiles。

• entryDate：FlowFile进入系统的日期和时间（即已创建）。此属性的值是一个数字，表示自1970年1月1日午夜（UTC）以来的毫秒数。

• lineageStartDate：任何时候克隆，合并或拆分FlowFile，都会导致创建“子”FlowFile。随着这些孩子被克隆，合并或拆分，建立了一系列祖先。该值表示最早的祖先进入系统的日期和时间。另一种思考方式是，此属性表示FlowFile通过系统的延迟。该值是一个数字，表示自1970年1月1日午夜（UTC）以来的毫秒数。

• fileSize：此属性表示FlowFile内容占用的字节数。

需要注意的是uuid，entryDate，lineageStartDate，和fileSize属性是系统生成的，不能改变。

提取属性

NiFi提供了几种不同的处理器，用于从FlowFiles中提取属性。可以在上面的“ 属性提取”部分中找到用于此目的的常用处理器列表。这是构建自定义处理器的一个非常常见的用例。编写许多处理器是为了理解特定的数据格式，并从FlowFile的内容中提取相关信息，创建属性来保存该信息，以便可以决定如何路由或处理数据。

添加用户定义的属性

除了能够将特定信息片段从FlowFile内容提取到属性中的处理器之外，用户还希望将自己的用户定义属性添加到流中特定位置的每个FlowFile。UpdateAttribute Processor专为此目的而设计。用户可以通过单击“属性”选项卡右上角的“+”按钮，在“配置”对话框中向处理器添加新属性。然后提示用户输入属性的名称，然后输入值。对于此UpdateAttribute Processor处理的每个FlowFile，将为每个用户定义的属性添加一个Attribute。Attribute的名称将与添加的属性的名称相同。

属性的值也可以包含表达式语言。这允许基于其他属性修改或添加属性。例如，如果我们想要将处理文件的主机名和日期添加到文件名之前，我们可以通过添加名称filename和值的属性来实现${hostname()}-${now():format(‘yyyy-dd-MM‘)}-${filename}。虽然这一开始可能会让人感到困惑，但下面关于表达式语言/使用属性值中的属性的部分将有助于清理这里发生的事情。

除了始终添加一组已定义的属性外，UpdateAttribute Processor还具有一个高级UI，允许用户配置一组规则，以便在何时添加属性。要访问此功能，请在“配置”对话框的“属性”选项卡中，单击Advanced对话框底部的按钮。这将提供专门为此处理器定制的UI，而不是为所有处理器提供的简单属性表。在此UI中，用户可以配置规则引擎，实质上是指定必须匹配的规则，以便将已配置的属性添加到FlowFile。

属性路由

NiFi最强大的功能之一是能够根据属性路由FlowFiles。执行此操作的主要机制是RouteOnAttribute Processor。此处理器与UpdateAttribute一样，通过添加用户定义的属性进行配置。通过单击处理器的“配置”对话框中“属性”选项卡右上角的“+”按钮，可以添加任意数量的属性。

每个FlowFile的属性将与配置的属性进行比较，以确定FlowFile是否满足指定的条件。每个属性的值应该是一个表达式语言表达式并返回一个布尔值。有关表达式语言的更多信息，请参阅下面的“ 属性值中的表达式语言/使用属性”部分。

在评估针对FlowFile的属性提供的表达式语言表达式之后，处理器根据所选的路由策略确定如何路由FlowFile。最常见的策略是“物业名称路径”策略。选择此策略后，处理器将为配置的每个属性公开关系。如果FlowFile的属性满足给定的表达式，则FlowFile的副本将路由到相应的Relationship。例如，如果我们有一个名为“begin-with-r”的新属性和值“$ {filename：startsWith（‘r‘）}”，那么任何文件名以字母‘r‘开头的FlowFile将是路由到那个关系。所有其他FlowFiles将被路由到“无与伦比”。

表达式语言/在属性值中使用属性

当我们从FlowFiles的内容中提取属性并添加用户定义的属性时，除非我们有一些可以使用它们的机制，否则它们不会作为运算符做得很好。NiFi表达式语言允许我们在配置流时访问和操作FlowFile属性值。并非所有处理器属性都允许使用表达式语言，但很多都可以。为了确定属性是否支持表达式语言，用户可以将鼠标悬停在 “处理器配置”对话框的“属性”选项卡中的“帮助”图标上 。这将提供一个工具提示，显示属性的描述，默认值（如果有）以及属性是否支持表达式语言。

对于支持表达式语言的属性，可以通过在开始${标记和结束 }标记中添加表达式来使用它。表达式可以像属性名一样简单。例如，要引用uuidAttribute，我们可以简单地使用该值${uuid}。如果属性名称以字母以外的任何字符开头，或者包含除数字，字母，句点（。）或下划线（_）以外的字符，则需要引用属性名称。例如，${My Attribute Name} 将无效，但${‘My Attribute Name‘}将引用属性My Attribute Name。

除了引用属性值之外，我们还可以对这些属性执行许多功能和比较。例如，如果我们想检查filename属性是否包含字母‘r‘而不注意大小写（大写或小写），我们可以使用表达式来完成${filename:toLower():contains(‘r‘)}。请注意，函数由冒号分隔。我们可以将任意数量的函数链接在一起以构建更复杂的表达式。同样重要的是要明白，即使我们正在调用filename:toLower()，这也不会改变filename属性的价值，而只是给我们一个新的价值。

我们也可以在另一个中嵌入一个表达式。例如，如果我们想要将attr1Attribute 的值与Attribute的值进行比较attr2，我们可以使用以下表达式来执行此操作：${attr1:equals( ${attr2} )}。

表达式语言包含许多不同的函数，可用于执行路由和操作属性所需的任务。存在用于解析和操作字符串，比较字符串和数值，操纵和替换值以及比较值的函数。对可用的不同功能的完整解释超出了本文档的范围，但“ 表达式语言指南”为每个功能提供了更多详细信息。

此外，此表达式语言指南内置于应用程序中，以便用户可以轻松查看哪些功能可用，并在键入时查看其文档。设置支持表达式语言的属性的值时，如果光标位于表达式语言的开始和结束标记内，则在关键字上按Ctrl + Space将弹出所有可用的函数，并将提供自动 - 完整的功能。单击或使用键盘导航到弹出窗口中列出的某个功能将导致显示工具提示，这解释了该功能的作用，它所期望的参数以及函数的返回类型。

表达式语言中的自定义属性

除了使用FlowFile属性外，您还可以定义表达式语言使用的自定义属性。定义自定义属性为处理和配置数据流提供了额外的灵活性。例如，您可以引用连接，服务器和服务属性的自定义属性。创建自定义属性后，您可以nifi.variable.registry.properties在“nifi.properties”文件中的字段中标识它们的位置。更新‘nifi.properties‘文件并重新启动NiFi后，您可以根据需要使用自定义属性。

使用模板

当我们使用处理器在NiFi中构建越来越复杂的数据流时，我们经常会发现我们将相同的处理器序列串在一起以执行某些任务。这可能变得乏味且低效。为解决这个问题，NiFi提供了模板概念。模板可以被认为是可重用的子流。要创建模板，请按照下列步骤操作：

• 选择要包含在模板中的组件。我们可以通过单击第一个组件选择多个组件，然后按住Shift键同时选择其他组件（以包括这些组件之间的连接），或者在画布上拖动所需组件周围的框时按住Shift键。

•  从“操作”调板中选择“创建模板图标”（ ）。

• 提供模板的名称和可选的描述。

• 单击Create按钮。

一旦我们创建了一个模板，我们就可以将它用作流程中的构建块，就像处理器一样。为此，我们将单击并将“模板”图标（ ）从“组件”工具栏拖动 到我们的画布上。然后，我们可以选择要添加到画布的模板，然后单击Add按钮。

最后，我们可以使用“模板管理”对话框来管理模板。要访问此对话框，请从全局菜单中选择模板。从这里，我们可以看到存在哪些模板并过滤模板以找到感兴趣的模板。在表的右侧是一个图标，用于将模板导出或下载为XML文件。然后可以将其提供给其他人，以便他们可以使用您的模板。

要将模板导入NiFi实例，请 从“操作”选项板中选择“上载模板”图标（ ），单击“搜索”图标并导航到计算机上的文件。然后单击Upload按钮。模板现在将显示在您的表格中，您可以将其拖动到画布上，就像您创建的任何其他模板一样。

使用模板时需要记住一些重要注意事项：

• 任何标识为敏感属性的属性（例如在处理器中配置的密码）都不会添加到模板中。每次将模板添加到画布时，都必须填充这些敏感属性。

• 如果模板中包含的组件引用Controller Service，则Controller Service也将添加到模板中。这意味着每次将模板添加到图表时，它都会创建Controller Service的副本。

监控NiFi

当数据在NiFi中流经您的数据流时，了解您的系统执行情况以评估您是否需要更多资源以及评估当前资源的运行状况非常重要。NiFi提供了一些监控系统的机制。

状态栏

在“组件”工具栏下的NiFi屏幕顶部附近有一个条形，称为状态栏。它包含一些关于NiFi当前健康状况的重要统计数据。活动线程的数量可以指示NiFi当前的工作状态，排队统计数据表示当前在整个流程中排队的FlowFile数量以及这些FlowFiles的总大小。

如果NiFi实例位于群集中，我们还会在此处看到一个指示器，告诉我们群集中有多少节点以及当前连接的节点数。在这种情况下，活动线程的数量和队列大小指示当前连接的所有节点的所有总和。

组件统计

画布上的每个处理器，进程组和远程进程组都提供了有关组件处理了多少数据的若干统计信息。这些统计信息提供有关在过去五分钟内处理了多少数据的信息。这是一个滚动窗口，允许我们查看处理器消耗的FlowFiles数量，以及处理器发出的FlowFiles数量。

处理器之间的连接还会显示当前排队的项目数。

查看这些指标的历史值以及（如果是群集的）不同节点如何相互比较也可能很有价值。为了查看此信息，我们可以右键单击组件并选择Stats菜单项。这将向我们展示一个图表，该图表涵盖自NiFi启动以来的时间，或最多24小时，以较少者为准。通过更改属性文件中的配置，可以扩展或减少此处显示的时间量。

在此对话框的右上角有一个下拉列表，允许用户选择他们正在查看的指标。底部的图表允许用户选择图表的较小部分进行放大。

公告

除了每个组件提供的统计信息之外，用户还想知道是否出现任何问题。虽然我们可以监视日志中的任何有趣内容，但在屏幕上弹出通知会更方便。如果处理器将任何内容记录为WARNING或ERROR，我们将在处理器的右上角看到“Bulletin Indicator”。此指示器看起来像一个粘滞便笺，将在事件发生后显示五分钟。将鼠标悬停在公告上可提供有关所发生情况的信息，以便用户无需筛选日志消息即可查找。如果在群集中，公告还将指示群集中的哪个节点发布公告。我们还可以在处理器的“配置”对话框的“设置”选项卡中更改公告的日志级别。

如果框架发布公告，我们还会在屏幕右上方突出显示公告指示符。在全局菜单中是公告板选项。单击此选项将我们带到公告板，在那里我们可以看到NiFi实例中出现的所有公告，并可以根据组件，消息等进行过滤。

数据来源

NiFi对其摄取的每个数据保持非常精细的细节。当数据通过系统处理并被转换，路由，拆分，聚合和分发到其他端点时，这些信息都存储在NiFi的Provenance Repository中。为了搜索和查看此信息，我们可以从全局菜单中选择数据源。这将为我们提供一个表格，列出我们搜索过的Provenance事件：

最初，此表填充了最近发生的1,000个Provenance事件（尽管事件发生后可能需要几秒钟才能处理信息）。在此对话框中，有一个Search按钮，允许用户搜索特定处理器发生的事件，按文件名或UUID或其他几个字段搜索特定的FlowFile。该nifi.properties文件提供了配置这些属性中的哪些属性编入索引或可搜索的功能。此外，属性文件还允许您选择将被索引的特定FlowFile属性。因此，您可以选择哪些属性对您的特定数据流很重要，并使这些属性可搜索。

活动详情

一旦我们执行了搜索，我们的表格将仅填充与搜索条件匹配的事件。从这里，我们可以选择 表左侧的Info图标（ ）来查看该事件的详细信息：

从这里，我们可以确切地看到该事件发生的时间，事件影响的FlowFile，事件执行的组件（处理器等），事件花费的时间以及事件发生时NiFi数据的总体时间发生了（总潜伏期）。

下一个选项卡提供事件发生时FlowFile上存在的所有属性的列表：

从这里，我们可以看到事件发生时FlowFile上存在的所有属性，以及这些属性的先前值。这允许我们知道哪些属性因此事件而发生变化以及它们如何变化。此外，在右侧角是一个复选框，允许用户仅查看那些已更改的属性。如果FlowFile只有少数属性，这可能不是特别有用，但当FlowFile有数百个属性时，它可能非常有用。

这非常重要，因为它允许用户理解FlowFile处理的确切上下文。理解FlowFile的“原因”是有帮助的，特别是在使用表达式语言配置处理器时。

最后，我们有Content选项卡：

此选项卡向我们提供有关存储FlowFile内容的内容存储库位置的信息。如果事件修改了FlowFile的内容，我们将看到‘之前‘（输入）和‘之后‘（输出）内容声明。如果数据格式是NiFi了解如何呈现的数据格式，我们可以选择下载内容或查看NiFi内部的内容。

此外，在选项卡的重播部分，有一个“重播”按钮，允许用户将FlowFile重新插入到流中，并从事件发生的时间点重新处理它。这提供了一种非常强大的机制，因为我们能够实时修改流程，重新处理FlowFile，然后查看结果。如果它们不符合预期，我们可以再次修改流程，并再次重新处理FlowFile。我们能够执行流程的迭代开发，直到它完全按照预期处理数据。

谱系图

除了查看Provenance事件的详细信息之外，我们还可以通过单击 表视图中的Lineage图标（）来查看所涉及的FlowFile的谱系。

这为我们提供了一个图形表示，说明了在遍历系统时该数据发生了什么：

从这里，我们可以右键单击任何表示的事件，然后单击View Details菜单项以查看事件详细信息。此图形表示向我们准确显示了数据发生的事件。有一些“特殊”事件类型需要注意。如果我们看到JOIN，FORK或CLONE事件，我们可以右键单击并选择Find Parents或Expand。这允许我们查看父FlowFiles和创建的子FlowFiles的谱系。

左下角的滑块允许我们查看这些事件发生的时间。通过左右滑动，我们可以看到哪些事件将延迟引入系统，这样我们就可以非常好地了解系统中可能需要提供更多资源的位置，例如处理器的并发任务数。或者它可能揭示，例如，大多数延迟是由JOIN事件引入的，我们在等待更多的FlowFiles连接在一起。在任何一种情况下，能够轻松查看其发生的位置是一项非常强大的功能，可帮助用户了解企业的??运营方式。

哪里有更多信息

NiFi社区已经建立了大量关于如何使用该软件的文档。除本入门指南外，还提供以下指南：

• Apache NiFi概述 - 概述Apache NiFi是什么，它做什么，以及它的创建原因。

• Apache NiFi用户指南 - 一个相当广泛的指南，经常被用作参考指南，因为它对构成应用程序的所有不同组件进行了相当冗长的讨论。本指南以NiFi运营商为其受众编写。它提供有关NiFi中可用的每个不同组件的信息，并说明如何使用应用程序提供的不同功能。

• 管理指南 - 为生产环境设置和管理Apache NiFi的指南。本指南提供有关不同系统级设置的信息，例如设置NiFi群集以及保护对Web UI和数据的访问。

• 表达语言指南 - 比上面提供的更为详尽的理解表达式语言的指南。本指南是NiFi表达语言的权威文档。它提供了EL的介绍以及每个函数，其参数和返回类型的解释以及提供示例。

• 开发人员指南 - 虽然不是All Things NiFi Development的详尽指南，但本指南提供了有关不同API的可用性以及如何使用它们的全面概述。此外，它还提供了开发NiFi组件和常用处理器习语的最佳实践，以帮助理解许多现有NiFi组件背后的逻辑。

• 贡献者指南 - 解释如何将工作贡献回Apache NiFi社区以便其他人可以使用它的指南。

Apache NiFi博客网站上还添加了几篇博客文章：https：//blogs.apache.org/nifi/

除了此处提供的博客和指南，您还可以浏览不同的 NiFi邮件列表，或发送电子邮件至[e-mail]或[e-mail]之一的邮件列表 。

NiFi社区的许多成员也可以通过Twitter获得，并积极监控提及@apachenifi的推文。



原文: http://nifi.apache.org/docs.html